Skip to content

Commit bcae32f

Browse files
semaphore-agent-production[bot]semaphore-agent-production[bot]
committed
Merge branch '8.2.x' into 8.3.x by rayokota
2 parents c969c75 + b2e4bb0 commit bcae32f

2 files changed

Lines changed: 181 additions & 14 deletions

File tree

‎protobuf-converter/src/main/java/io/confluent/connect/protobuf/ProtobufData.java‎

Lines changed: 49 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,11 @@ public class ProtobufData {
103103
public static final String PROTOBUF_TYPE_UNION_PREFIX = PROTOBUF_TYPE_UNION + ".";
104104
public static final String PROTOBUF_TYPE_TAG = NAMESPACE + ".Tag";
105105
public static final String PROTOBUF_TYPE_PROP = NAMESPACE + ".Type";
106+
// Marks an optional scalar that originated from a protobuf wrapper type
107+
// (e.g. google.protobuf.StringValue). Used to distinguish a genuine wrapper-typed oneof
108+
// member from a plain scalar oneof member: oneof members are always optional, so optionality
109+
// alone cannot signal wrapper-ness for them the way it does for ordinary fields.
110+
public static final String PROTOBUF_TYPE_WRAPPER = NAMESPACE + ".Wrapper";
106111

107112
public static final String PROTOBUF_PRECISION_PROP = "precision";
108113
public static final String PROTOBUF_SCALE_PROP = "scale";
@@ -403,10 +408,12 @@ private Object fromConnectData(
403408
}
404409
}
405410

406-
// Oneof (union) members express nullability via the oneof itself, not via wrapper types,
407-
// so they must never be wrapped even when useWrapperForNullables is set.
411+
// A plain scalar oneof member expresses nullability via the oneof itself, so it must not be
412+
// wrapped even when useWrapperForNullables is set. A genuine wrapper-typed oneof member
413+
// (carrying the wrapper marker) must still be wrapped so it round-trips faithfully.
408414
boolean isWrapper = isWrapper(protobufSchema)
409-
|| (useWrapperForNullables && !oneofMember && schema.isOptional());
415+
|| (useWrapperForNullables && schema.isOptional()
416+
&& (!oneofMember || hasWrapperMarker(schema)));
410417
final Schema.Type schemaType = schema.type();
411418
try {
412419
switch (schemaType) {
@@ -603,6 +610,13 @@ private boolean isWrapper(ProtobufSchema protobufSchema) {
603610
}
604611
}
605612

613+
// True if the schema is an (optional scalar) oneof member that originated from a protobuf
614+
// wrapper type and must therefore be re-wrapped on the write side.
615+
private boolean hasWrapperMarker(Schema schema) {
616+
return schema.parameters() != null
617+
&& Boolean.parseBoolean(schema.parameters().get(PROTOBUF_TYPE_WRAPPER));
618+
}
619+
606620
private Object getFieldType(Object ctx, String name) {
607621
FieldDescriptor field = ((Descriptor) ctx).findFieldByName(name);
608622
if (field == null) {
@@ -1018,11 +1032,13 @@ private String dataTypeFromConnectSchema(
10181032
switch (schema.type()) {
10191033
case INT8:
10201034
params.put(CONNECT_TYPE_PROP, CONNECT_TYPE_INT8);
1021-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1035+
return useWrapperForNullables && schema.isOptional()
1036+
&& (!oneofMember || hasWrapperMarker(schema))
10221037
? PROTOBUF_INT32_WRAPPER_TYPE : FieldDescriptor.Type.INT32.toString().toLowerCase();
10231038
case INT16:
10241039
params.put(CONNECT_TYPE_PROP, CONNECT_TYPE_INT16);
1025-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1040+
return useWrapperForNullables && schema.isOptional()
1041+
&& (!oneofMember || hasWrapperMarker(schema))
10261042
? PROTOBUF_INT32_WRAPPER_TYPE : FieldDescriptor.Type.INT32.toString().toLowerCase();
10271043
case INT32:
10281044
if (schema.parameters() != null && schema.parameters().containsKey(PROTOBUF_TYPE_ENUM)) {
@@ -1032,7 +1048,8 @@ private String dataTypeFromConnectSchema(
10321048
if (schema.parameters() != null && schema.parameters().containsKey(PROTOBUF_TYPE_PROP)) {
10331049
defaultType = schema.parameters().get(PROTOBUF_TYPE_PROP);
10341050
}
1035-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1051+
return useWrapperForNullables && schema.isOptional()
1052+
&& (!oneofMember || hasWrapperMarker(schema))
10361053
? PROTOBUF_INT32_WRAPPER_TYPE : defaultType;
10371054
case INT64:
10381055
defaultType = FieldDescriptor.Type.INT64.toString().toLowerCase();
@@ -1052,16 +1069,20 @@ private String dataTypeFromConnectSchema(
10521069
default:
10531070
wrapperType = PROTOBUF_INT64_WRAPPER_TYPE;
10541071
}
1055-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1072+
return useWrapperForNullables && schema.isOptional()
1073+
&& (!oneofMember || hasWrapperMarker(schema))
10561074
? wrapperType : defaultType;
10571075
case FLOAT32:
1058-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1076+
return useWrapperForNullables && schema.isOptional()
1077+
&& (!oneofMember || hasWrapperMarker(schema))
10591078
? PROTOBUF_FLOAT_WRAPPER_TYPE : FieldDescriptor.Type.FLOAT.toString().toLowerCase();
10601079
case FLOAT64:
1061-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1080+
return useWrapperForNullables && schema.isOptional()
1081+
&& (!oneofMember || hasWrapperMarker(schema))
10621082
? PROTOBUF_DOUBLE_WRAPPER_TYPE : FieldDescriptor.Type.DOUBLE.toString().toLowerCase();
10631083
case BOOLEAN:
1064-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1084+
return useWrapperForNullables && schema.isOptional()
1085+
&& (!oneofMember || hasWrapperMarker(schema))
10651086
? PROTOBUF_BOOL_WRAPPER_TYPE : FieldDescriptor.Type.BOOL.toString().toLowerCase();
10661087
case STRING:
10671088
if (schema.parameters() != null) {
@@ -1071,10 +1092,12 @@ private String dataTypeFromConnectSchema(
10711092
return schema.parameters().get(PROTOBUF_TYPE_ENUM);
10721093
}
10731094
}
1074-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1095+
return useWrapperForNullables && schema.isOptional()
1096+
&& (!oneofMember || hasWrapperMarker(schema))
10751097
? PROTOBUF_STRING_WRAPPER_TYPE : FieldDescriptor.Type.STRING.toString().toLowerCase();
10761098
case BYTES:
1077-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1099+
return useWrapperForNullables && schema.isOptional()
1100+
&& (!oneofMember || hasWrapperMarker(schema))
10781101
? PROTOBUF_BYTES_WRAPPER_TYPE : FieldDescriptor.Type.BYTES.toString().toLowerCase();
10791102
case ARRAY:
10801103
// Array should not occur here
@@ -1451,7 +1474,7 @@ private Schema toConnectSchema(ToConnectContext ctx, FieldDescriptor descriptor)
14511474
}
14521475

14531476
private Schema toConnectSchema(
1454-
ToConnectContext ctx, FieldDescriptor descriptor, boolean forceOptional) {
1477+
ToConnectContext ctx, FieldDescriptor descriptor, boolean oneofMember) {
14551478
SchemaBuilder builder;
14561479

14571480
switch (descriptor.getType()) {
@@ -1593,10 +1616,18 @@ private Schema toConnectSchema(
15931616
builder.optional();
15941617
}
15951618

1596-
if (forceOptional) {
1619+
if (oneofMember) {
15971620
// Union (oneof) members are inherently nullable, since at most one is ever set,
15981621
// so they must be optional regardless of the nullable handling configs.
15991622
builder.optional();
1623+
if (useWrapperForNullables
1624+
&& descriptor.getType() == FieldDescriptor.Type.MESSAGE
1625+
&& isWrapperType(descriptor.getMessageType())) {
1626+
// The member was a genuine wrapper type that got unwrapped to an optional scalar.
1627+
// Since oneof members are always optional, record this so the write side can re-wrap
1628+
// it (a plain scalar member, which must stay unwrapped, looks identical otherwise).
1629+
builder.parameter(PROTOBUF_TYPE_WRAPPER, Boolean.TRUE.toString());
1630+
}
16001631
} else if (useOptionalForNullables) {
16011632
if (hasOptionalKeyword(descriptor)) {
16021633
builder.optional();
@@ -1608,6 +1639,10 @@ private Schema toConnectSchema(
16081639
return builder.build();
16091640
}
16101641

1642+
private boolean isWrapperType(Descriptor descriptor) {
1643+
return toUnwrappedSchema(descriptor) != null;
1644+
}
1645+
16111646
private SchemaBuilder toUnwrappedOrStructSchema(
16121647
ToConnectContext ctx, FieldDescriptor descriptor) {
16131648
if (!useWrapperForNullables) {

‎protobuf-converter/src/test/java/io/confluent/connect/protobuf/ProtobufDataTest.java‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@
9797
import static io.confluent.connect.protobuf.ProtobufData.PROTOBUF_TYPE_PROP;
9898
import static io.confluent.connect.protobuf.ProtobufData.PROTOBUF_TYPE_TAG;
9999
import static io.confluent.connect.protobuf.ProtobufData.PROTOBUF_TYPE_UNION_PREFIX;
100+
import static io.confluent.connect.protobuf.ProtobufData.PROTOBUF_TYPE_WRAPPER;
100101
import static io.confluent.kafka.serializers.protobuf.test.TimestampValueOuterClass.TimestampValue.newBuilder;
101102
import static org.junit.Assert.assertArrayEquals;
102103
import static org.junit.Assert.assertEquals;
@@ -2436,6 +2437,10 @@ private void assertOneofBranchRoundTrips(
24362437
Struct members = flatten ? value : (Struct) value.get("payload");
24372438
assertEquals(payloadValue, members.get(payloadField));
24382439

2440+
// Write-side fidelity is only asserted for the (non-flattened) union representation. With
2441+
// flattenUnions the oneof grouping is lost on read (members become plain top-level fields),
2442+
// so the converter cannot reconstruct the oneof on write and a byte-faithful round-trip is
2443+
// not achievable by design; these flattened cases therefore verify the read side only.
24392444
if (!flatten) {
24402445
// Write the Connect data back to protobuf. The regenerated schema must keep the oneof
24412446
// member as its original scalar type (not a google.protobuf wrapper message), and the
@@ -2449,6 +2454,133 @@ private void assertOneofBranchRoundTrips(
24492454
}
24502455
}
24512456

2457+
private static final String WRAPPER_ONEOF_SCHEMA = "syntax = \"proto3\";\n"
2458+
+ "package io.confluent.test;\n"
2459+
+ "import \"google/protobuf/wrappers.proto\";\n"
2460+
+ "message WrapperOneofMessage {\n"
2461+
+ " google.protobuf.StringValue display_name = 2;\n" // wrapper, NOT in oneof
2462+
+ " google.protobuf.BoolValue verified = 3;\n" // wrapper, NOT in oneof
2463+
+ " oneof payment_method {\n"
2464+
+ " google.protobuf.StringValue credit_card = 7;\n" // wrapper IN oneof
2465+
+ " google.protobuf.StringValue bank_account = 8;\n" // wrapper IN oneof
2466+
+ " }\n"
2467+
+ " oneof contact_info {\n"
2468+
+ " string email = 5;\n" // plain string IN oneof
2469+
+ " string phone = 6;\n" // plain string IN oneof
2470+
+ " }\n"
2471+
+ " oneof mixed_choice {\n"
2472+
+ " google.protobuf.Int32Value count = 9;\n" // wrapper IN oneof
2473+
+ " string note = 10;\n" // plain string IN oneof
2474+
+ " }\n"
2475+
+ "}\n";
2476+
2477+
@Test
2478+
public void testWrapperTypedOneofMembersRoundTrip() throws Exception {
2479+
ProtobufDataConfig config = new ProtobufDataConfig.Builder()
2480+
.with(ProtobufDataConfig.WRAPPER_FOR_NULLABLES_CONFIG, true)
2481+
.with(ProtobufDataConfig.GENERATE_INDEX_FOR_UNIONS_CONFIG, false)
2482+
.build();
2483+
ProtobufData protobufData = new ProtobufData(config);
2484+
ProtobufSchema protobufSchema = new ProtobufSchema(WRAPPER_ONEOF_SCHEMA);
2485+
Descriptor descriptor = protobufSchema.toDescriptor();
2486+
2487+
// Read side: wrapper types are unwrapped to optional scalars whether or not they are in a
2488+
// oneof. A wrapper-typed oneof member additionally carries the wrapper marker, which is how
2489+
// the write side tells it apart from a plain scalar member (both are optional scalars).
2490+
Schema connectSchema = protobufData.toConnectSchema(protobufSchema);
2491+
Schema displayName = connectSchema.field("display_name").schema();
2492+
assertEquals(Schema.Type.STRING, displayName.type());
2493+
assertTrue(displayName.isOptional());
2494+
assertNull(displayName.parameters().get(PROTOBUF_TYPE_WRAPPER)); // not a oneof member
2495+
2496+
Schema creditCard = connectSchema.field("payment_method").schema().field("credit_card").schema();
2497+
assertEquals(Schema.Type.STRING, creditCard.type());
2498+
assertEquals("true", creditCard.parameters().get(PROTOBUF_TYPE_WRAPPER)); // wrapper in oneof
2499+
2500+
Schema email = connectSchema.field("contact_info").schema().field("email").schema();
2501+
assertEquals(Schema.Type.STRING, email.type());
2502+
assertNull(email.parameters().get(PROTOBUF_TYPE_WRAPPER)); // plain scalar in oneof
2503+
2504+
// A single oneof mixing a wrapper-typed member and a plain member: the marker is per-member,
2505+
// so only the wrapper member carries it.
2506+
Schema mixed = connectSchema.field("mixed_choice").schema();
2507+
assertEquals(Schema.Type.INT32, mixed.field("count").schema().type());
2508+
assertEquals("true", mixed.field("count").schema().parameters().get(PROTOBUF_TYPE_WRAPPER));
2509+
assertEquals(Schema.Type.STRING, mixed.field("note").schema().type());
2510+
assertNull(mixed.field("note").schema().parameters().get(PROTOBUF_TYPE_WRAPPER));
2511+
2512+
// Each oneof branch must round-trip byte-for-byte: a wrapper member stays a wrapper message
2513+
// and a plain member stays a plain scalar.
2514+
assertWrapperOneofRoundTrips(protobufData, protobufSchema, descriptor, "credit_card", "email");
2515+
assertWrapperOneofRoundTrips(protobufData, protobufSchema, descriptor, "bank_account", "phone");
2516+
2517+
// Same, but for the two members of the single mixed oneof.
2518+
FieldDescriptor countField = descriptor.findFieldByName("count");
2519+
Descriptor int32ValueDesc = countField.getMessageType();
2520+
DynamicMessage countValue = DynamicMessage.newBuilder(int32ValueDesc)
2521+
.setField(int32ValueDesc.findFieldByName("value"), 5)
2522+
.build();
2523+
assertOneofMemberRoundTrips(protobufData, protobufSchema, descriptor,
2524+
"count", countValue, FieldDescriptor.Type.MESSAGE);
2525+
assertOneofMemberRoundTrips(protobufData, protobufSchema, descriptor,
2526+
"note", "hello", FieldDescriptor.Type.STRING);
2527+
}
2528+
2529+
private void assertOneofMemberRoundTrips(
2530+
ProtobufData protobufData,
2531+
ProtobufSchema protobufSchema,
2532+
Descriptor descriptor,
2533+
String field,
2534+
Object value,
2535+
FieldDescriptor.Type expectedType) {
2536+
DynamicMessage message = DynamicMessage.newBuilder(descriptor)
2537+
.setField(descriptor.findFieldByName(field), value)
2538+
.build();
2539+
byte[] originalBytes = message.toByteArray();
2540+
2541+
SchemaAndValue schemaAndValue = protobufData.toConnectData(protobufSchema, message);
2542+
ConnectSchema.validateValue(schemaAndValue.schema(), schemaAndValue.value());
2543+
2544+
ProtobufSchemaAndValue back =
2545+
protobufData.fromConnectData(schemaAndValue.schema(), schemaAndValue.value());
2546+
Descriptor regenerated = back.getSchema().toDescriptor();
2547+
assertEquals(expectedType, regenerated.findFieldByName(field).getType());
2548+
assertArrayEquals(originalBytes, ((Message) back.getValue()).toByteArray());
2549+
}
2550+
2551+
private void assertWrapperOneofRoundTrips(
2552+
ProtobufData protobufData,
2553+
ProtobufSchema protobufSchema,
2554+
Descriptor descriptor,
2555+
String paymentField,
2556+
String contactField) {
2557+
FieldDescriptor displayNameField = descriptor.findFieldByName("display_name");
2558+
Descriptor stringValueDesc = displayNameField.getMessageType();
2559+
DynamicMessage stringValue = DynamicMessage.newBuilder(stringValueDesc)
2560+
.setField(stringValueDesc.findFieldByName("value"), "shopper")
2561+
.build();
2562+
DynamicMessage paymentValue = DynamicMessage.newBuilder(stringValueDesc)
2563+
.setField(stringValueDesc.findFieldByName("value"), "secret")
2564+
.build();
2565+
DynamicMessage message = DynamicMessage.newBuilder(descriptor)
2566+
.setField(displayNameField, stringValue) // non-oneof wrapper
2567+
.setField(descriptor.findFieldByName(paymentField), paymentValue) // wrapper in oneof
2568+
.setField(descriptor.findFieldByName(contactField), "a@b.com") // plain string in oneof
2569+
.build();
2570+
byte[] originalBytes = message.toByteArray();
2571+
2572+
SchemaAndValue schemaAndValue = protobufData.toConnectData(protobufSchema, message);
2573+
ConnectSchema.validateValue(schemaAndValue.schema(), schemaAndValue.value());
2574+
2575+
ProtobufSchemaAndValue back =
2576+
protobufData.fromConnectData(schemaAndValue.schema(), schemaAndValue.value());
2577+
Descriptor regenerated = back.getSchema().toDescriptor();
2578+
// Wrapper members stay wrapper messages; plain members stay plain scalars.
2579+
assertEquals(FieldDescriptor.Type.MESSAGE, regenerated.findFieldByName(paymentField).getType());
2580+
assertEquals(FieldDescriptor.Type.STRING, regenerated.findFieldByName(contactField).getType());
2581+
assertArrayEquals(originalBytes, ((Message) back.getValue()).toByteArray());
2582+
}
2583+
24522584
@Test
24532585
public void testToConnectRecursiveSchema() {
24542586
ProtobufSchema protobufSchema = new ProtobufSchema(

0 commit comments

Comments
 (0)