Nested oneof

protobuf-kafkaconnect-converter/src/main/java/com/amazonaws/services/schemaregistry/kafkaconnect/protobuf/toconnectschema/ProtobufSchemaToConnectSchemaConverter.java

@@ -73,19 +73,23 @@ public Schema toConnectSchema(@NonNull final Message message) {
         builder.version(CONVERTER_VERSION);
         builder.parameter(PROTOBUF_PACKAGE, descriptor.getFile().getPackage());
 
+        processFieldDescriptors(builder, fieldDescriptorList);
+
+        return builder.build();
+    }
+
+    private void processFieldDescriptors(SchemaBuilder builder, List<Descriptors.FieldDescriptor> fieldDescriptorList) {
         for (final Descriptors.FieldDescriptor fieldDescriptor : fieldDescriptorList) {
             if (fieldDescriptor.getRealContainingOneof() != null) {
                 Descriptors.OneofDescriptor oneofDescriptor = fieldDescriptor.getRealContainingOneof();
-                if (!builder.fields().stream().anyMatch(field -> field.name().equals(oneofDescriptor.getName()))) {
+                if (builder.fields().stream().noneMatch(field -> field.name().equals(oneofDescriptor.getName()))) {
                     builder.field(oneofDescriptor.getName(), toConnectSchemaForOneOfField(oneofDescriptor));
                 }
-                continue;
+            } else {
+                final String fieldName = fieldDescriptor.getName();
+                builder.field(fieldName, toConnectSchemaForField(fieldDescriptor));
             }
-            final String fieldName = fieldDescriptor.getName();
-            builder.field(fieldName, toConnectSchemaForField(fieldDescriptor));
         }
-
-        return builder.build();
     }
 
     private Schema toConnectSchemaForField(final Descriptors.FieldDescriptor fieldDescriptor) {
@@ -204,9 +208,7 @@ private SchemaBuilder toConnectSchemaBuilderForField(final Descriptors.FieldDesc
 
                 String fullName = fieldDescriptor.getMessageType().getFullName();
                 schemaBuilder = SchemaBuilder.struct().name(fullName);
-                for (Descriptors.FieldDescriptor field : fieldDescriptor.getMessageType().getFields()) {
-                    schemaBuilder.field(field.getName(), toConnectSchemaForField(field));
-                }
+                processFieldDescriptors(schemaBuilder, fieldDescriptor.getMessageType().getFields());
                 break;
             }
             default:

protobuf-kafkaconnect-converter/src/test/java/com/amazonaws/services/schemaregistry/kafkaconnect/protobuf/ProtobufSchemaConverterTest.java

@@ -147,6 +147,9 @@ private static Stream<Arguments> getToConnectTestCases() {
             Arguments.of(ToConnectTestDataGenerator.getOneofProtobufMessages().get(0),
                 ToConnectTestDataGenerator.getOneofSchema(PACKAGE_NAME),
                 ToConnectTestDataGenerator.getOneofTypeData(PACKAGE_NAME)),
+            Arguments.of(ToConnectTestDataGenerator.getNestedOneofProtobufMessages().get(0),
+                ToConnectTestDataGenerator.getNestedOneOfSchema(PACKAGE_NAME),
+                ToConnectTestDataGenerator.getNestedOneOfTypeData(PACKAGE_NAME)),
             Arguments.of(ToConnectTestDataGenerator.getAllTypesProtobufMessages().get(0),
                 ToConnectTestDataGenerator.getAllTypesSchema(PACKAGE_NAME),
                 ToConnectTestDataGenerator.getAllTypesData(PACKAGE_NAME))
@@ -293,4 +296,4 @@ public void testSchemaCache_fromConnectConversion() {
         protobufSchemaConverter.fromConnectData(TOPIC_NAME, connectEnumSchema, connectEnumData);
         assertEquals(2, fromConnectSchemaCache.size());
     }
-}
+}

protobuf-kafkaconnect-converter/src/test/java/com/amazonaws/services/schemaregistry/kafkaconnect/protobuf/ToConnectTestDataGenerator.java

@@ -26,6 +26,7 @@
 import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.ArrayTypeSyntax3;
 import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.MapTypeSyntax3;
 import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.NestedTypeSyntax3;
+import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.NestedOneofTypeSyntax3;
 import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.OneofTypeSyntax3;
 import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.PrimitiveTypesSyntax3;
 import com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax2.EnumTypeSyntax2;
@@ -619,7 +620,10 @@ public static Struct getStructTypeData(String packageName) {
         final Struct connectData = new Struct(connectSchema);
 
         connectData
-                .put("address", new Struct(connectSchema.field("address").schema()).put("street", "8th").put("zipcode", 98121))
+                .put("address",
+                        new Struct(connectSchema.field("address").schema())
+                                .put("street", "8th")
+                                .put("zipcode", 98121))
                 .put("status", "VALID")
                 .put("customer", new Struct(connectSchema.field("customer").schema()).put("name", "joe"))
                 .put("mapping", ImmutableMap.of("hello", true))
@@ -702,6 +706,62 @@ private static Map<String, Schema> getOneofType() {
                 .build();
     }
 
+    public static List<Message> getNestedOneofProtobufMessages() {
+        NestedOneofTypeSyntax3.Metadata metadata = NestedOneofTypeSyntax3.Metadata.newBuilder()
+                .setMetadataId("hi")
+                .setRegistered("yup")
+                .build();
+        return Arrays.asList(
+                NestedOneofTypeSyntax3.Event.newBuilder()
+                        .setMetadata(metadata)
+                        .setEventId("123123")
+                        .build()
+        );
+    }
+
+    public static Schema getNestedOneOfSchema(String packageName) {
+        return createConnectSchema(
+               "Event",
+               getNestedOneOfType(packageName),
+               ImmutableMap.of(PROTOBUF_PACKAGE, packageName)
+        );
+    }
+
+    public static Struct getNestedOneOfTypeData(String packageName) {
+        final Schema connectSchema = getNestedOneOfSchema(packageName);
+        final Struct connectData = new Struct(connectSchema);
+
+        connectData
+                .put("metadata",
+                        new Struct(connectSchema.field("metadata").schema())
+                        .put("metadata_id", "hi")
+                        .put("status",
+                                new Struct(connectSchema.field("metadata").schema().field("status").schema())
+                                .put("registered", "yup")
+                        )
+                )
+                .put("event_id", "123123");
+        return connectData;
+    }
+
+    private static Map<String, Schema> getNestedOneOfType(String packageName) {
+        final SchemaBuilder metadataBuilder = SchemaBuilder.struct().name(getFullName(packageName, "Metadata"))
+                .field("metadata_id", SchemaBuilder.string().parameter(PROTOBUF_TAG, "1").build())
+                .field("status", SchemaBuilder.struct()
+                        .name("status")
+                        .field("registered", SchemaBuilder.string().parameter(PROTOBUF_TAG, "2").optional().build())
+                        .field("unregistered", SchemaBuilder.string().parameter(PROTOBUF_TAG, "3").optional().build())
+                        .parameter("protobuf.type", "oneof")
+                        .optional()
+                        .build()
+                );
+
+        return ImmutableMap.<String, Schema> builder()
+                .put("metadata", metadataBuilder.parameter(PROTOBUF_TAG, "1").build())
+                .put("event_id", SchemaBuilder.string().parameter(PROTOBUF_TAG, "2").build())
+                .build();
+    }
+
     public static List<Message> getAllTypesProtobufMessages() {
 
         com.google.type.Date.Builder dateBuilder = com.google.type.Date.newBuilder();

protobuf-kafkaconnect-converter/src/test/proto/NestedOneofTypeSyntax3.proto

@@ -0,0 +1,16 @@
+syntax = "proto3";
+
+package com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3;
+
+message Metadata {
+  string metadata_id = 1;
+  oneof status {
+    string registered = 2;
+    string unregistered = 3;
+  }
+}
+
+message Event {
+  .com.amazonaws.services.schemaregistry.kafkaconnect.tests.syntax3.Metadata metadata = 1;
+  string event_id = 2;
+}