Skip to content

Commit 1f2c4ac

Browse files
semaphore-agent-production[bot]semaphore-agent-production[bot]
committed
Merge branch '8.3.x' into master by rayokota
2 parents 4a9d290 + bcae32f commit 1f2c4ac

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) {
@@ -609,6 +616,13 @@ private boolean isWrapper(ProtobufSchema protobufSchema) {
609616
}
610617
}
611618

619+
// True if the schema is an (optional scalar) oneof member that originated from a protobuf
620+
// wrapper type and must therefore be re-wrapped on the write side.
621+
private boolean hasWrapperMarker(Schema schema) {
622+
return schema.parameters() != null
623+
&& Boolean.parseBoolean(schema.parameters().get(PROTOBUF_TYPE_WRAPPER));
624+
}
625+
612626
private Object getFieldType(Object ctx, String name) {
613627
FieldDescriptor field = ((Descriptor) ctx).findFieldByName(name);
614628
if (field == null) {
@@ -1024,11 +1038,13 @@ private String dataTypeFromConnectSchema(
10241038
switch (schema.type()) {
10251039
case INT8:
10261040
params.put(CONNECT_TYPE_PROP, CONNECT_TYPE_INT8);
1027-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1041+
return useWrapperForNullables && schema.isOptional()
1042+
&& (!oneofMember || hasWrapperMarker(schema))
10281043
? PROTOBUF_INT32_WRAPPER_TYPE : FieldDescriptor.Type.INT32.toString().toLowerCase();
10291044
case INT16:
10301045
params.put(CONNECT_TYPE_PROP, CONNECT_TYPE_INT16);
1031-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1046+
return useWrapperForNullables && schema.isOptional()
1047+
&& (!oneofMember || hasWrapperMarker(schema))
10321048
? PROTOBUF_INT32_WRAPPER_TYPE : FieldDescriptor.Type.INT32.toString().toLowerCase();
10331049
case INT32:
10341050
if (schema.parameters() != null && schema.parameters().containsKey(PROTOBUF_TYPE_ENUM)) {
@@ -1038,7 +1054,8 @@ private String dataTypeFromConnectSchema(
10381054
if (schema.parameters() != null && schema.parameters().containsKey(PROTOBUF_TYPE_PROP)) {
10391055
defaultType = schema.parameters().get(PROTOBUF_TYPE_PROP);
10401056
}
1041-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1057+
return useWrapperForNullables && schema.isOptional()
1058+
&& (!oneofMember || hasWrapperMarker(schema))
10421059
? PROTOBUF_INT32_WRAPPER_TYPE : defaultType;
10431060
case INT64:
10441061
defaultType = FieldDescriptor.Type.INT64.toString().toLowerCase();
@@ -1058,16 +1075,20 @@ private String dataTypeFromConnectSchema(
10581075
default:
10591076
wrapperType = PROTOBUF_INT64_WRAPPER_TYPE;
10601077
}
1061-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1078+
return useWrapperForNullables && schema.isOptional()
1079+
&& (!oneofMember || hasWrapperMarker(schema))
10621080
? wrapperType : defaultType;
10631081
case FLOAT32:
1064-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1082+
return useWrapperForNullables && schema.isOptional()
1083+
&& (!oneofMember || hasWrapperMarker(schema))
10651084
? PROTOBUF_FLOAT_WRAPPER_TYPE : FieldDescriptor.Type.FLOAT.toString().toLowerCase();
10661085
case FLOAT64:
1067-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1086+
return useWrapperForNullables && schema.isOptional()
1087+
&& (!oneofMember || hasWrapperMarker(schema))
10681088
? PROTOBUF_DOUBLE_WRAPPER_TYPE : FieldDescriptor.Type.DOUBLE.toString().toLowerCase();
10691089
case BOOLEAN:
1070-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1090+
return useWrapperForNullables && schema.isOptional()
1091+
&& (!oneofMember || hasWrapperMarker(schema))
10711092
? PROTOBUF_BOOL_WRAPPER_TYPE : FieldDescriptor.Type.BOOL.toString().toLowerCase();
10721093
case STRING:
10731094
if (schema.parameters() != null) {
@@ -1077,10 +1098,12 @@ private String dataTypeFromConnectSchema(
10771098
return schema.parameters().get(PROTOBUF_TYPE_ENUM);
10781099
}
10791100
}
1080-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1101+
return useWrapperForNullables && schema.isOptional()
1102+
&& (!oneofMember || hasWrapperMarker(schema))
10811103
? PROTOBUF_STRING_WRAPPER_TYPE : FieldDescriptor.Type.STRING.toString().toLowerCase();
10821104
case BYTES:
1083-
return useWrapperForNullables && schema.isOptional() && !oneofMember
1105+
return useWrapperForNullables && schema.isOptional()
1106+
&& (!oneofMember || hasWrapperMarker(schema))
10841107
? PROTOBUF_BYTES_WRAPPER_TYPE : FieldDescriptor.Type.BYTES.toString().toLowerCase();
10851108
case ARRAY:
10861109
// Array should not occur here
@@ -1463,7 +1486,7 @@ private Schema toConnectSchema(ToConnectContext ctx, FieldDescriptor descriptor)
14631486
}
14641487

14651488
private Schema toConnectSchema(
1466-
ToConnectContext ctx, FieldDescriptor descriptor, boolean forceOptional) {
1489+
ToConnectContext ctx, FieldDescriptor descriptor, boolean oneofMember) {
14671490
SchemaBuilder builder;
14681491

14691492
switch (descriptor.getType()) {
@@ -1605,10 +1628,18 @@ private Schema toConnectSchema(
16051628
builder.optional();
16061629
}
16071630

1608-
if (forceOptional) {
1631+
if (oneofMember) {
16091632
// Union (oneof) members are inherently nullable, since at most one is ever set,
16101633
// so they must be optional regardless of the nullable handling configs.
16111634
builder.optional();
1635+
if (useWrapperForNullables
1636+
&& descriptor.getType() == FieldDescriptor.Type.MESSAGE
1637+
&& isWrapperType(descriptor.getMessageType())) {
1638+
// The member was a genuine wrapper type that got unwrapped to an optional scalar.
1639+
// Since oneof members are always optional, record this so the write side can re-wrap
1640+
// it (a plain scalar member, which must stay unwrapped, looks identical otherwise).
1641+
builder.parameter(PROTOBUF_TYPE_WRAPPER, Boolean.TRUE.toString());
1642+
}
16121643
} else if (useOptionalForNullables) {
16131644
if (hasOptionalKeyword(descriptor)) {
16141645
builder.optional();
@@ -1620,6 +1651,10 @@ private Schema toConnectSchema(
16201651
return builder.build();
16211652
}
16221653

1654+
private boolean isWrapperType(Descriptor descriptor) {
1655+
return toUnwrappedSchema(descriptor) != null;
1656+
}
1657+
16231658
private SchemaBuilder toUnwrappedOrStructSchema(
16241659
ToConnectContext ctx, FieldDescriptor descriptor) {
16251660
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)