DGS-6023 Add Protobuf converter config to not generate index for unions

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

@@ -304,6 +304,7 @@ public class ProtobufData {
   private boolean useWrapperForNullables;
   private boolean useWrapperForRawPrimitives;
   private boolean generateStructForNulls;
+  private boolean generateIndexForUnions;
 
   public ProtobufData() {
     this(new ProtobufDataConfig.Builder().with(
@@ -329,6 +330,7 @@ public ProtobufData(ProtobufDataConfig protobufDataConfig) {
     this.useWrapperForNullables = protobufDataConfig.useWrapperForNullables();
     this.useWrapperForRawPrimitives = protobufDataConfig.useWrapperForRawPrimitives();
     this.generateStructForNulls = protobufDataConfig.generateStructForNulls();
+    this.generateIndexForUnions = protobufDataConfig.generateIndexForUnions();
   }
 
   /**
@@ -1359,7 +1361,11 @@ private void setUnionField(
   }
 
   private String unionFieldName(OneofDescriptor oneofDescriptor) {
-    return oneofDescriptor.getName() + "_" + oneofDescriptor.getIndex();
+    String name = oneofDescriptor.getName();
+    if (generateIndexForUnions) {
+      name += "_" + oneofDescriptor.getIndex();
+    }
+    return name;
   }
 
   private void setStructField(

protobuf-converter/src/main/java/io/confluent/connect/protobuf/ProtobufDataConfig.java

@@ -64,6 +64,11 @@ public class ProtobufDataConfig extends AbstractDataConfig {
   public static final String GENERATE_STRUCT_FOR_NULLS_DOC = "Whether to generate a default struct "
       + "for null messages";
 
+  public static final String GENERATE_INDEX_FOR_UNIONS_CONFIG = "generate.index.for.unions";
+  public static final boolean GENERATE_INDEX_FOR_UNIONS_DEFAULT = true;
+  public static final String GENERATE_INDEX_FOR_UNIONS_DOC = "Whether to suffix union"
+      + "names with an underscore followed by an index";
+
   public static ConfigDef baseConfigDef() {
     return AbstractDataConfig.baseConfigDef()
         .define(ENHANCED_PROTOBUF_SCHEMA_SUPPORT_CONFIG,
@@ -102,7 +107,12 @@ public static ConfigDef baseConfigDef() {
             ConfigDef.Type.BOOLEAN,
             GENERATE_STRUCT_FOR_NULLS_DEFAULT,
             ConfigDef.Importance.MEDIUM,
-            GENERATE_STRUCT_FOR_NULLS_DOC
+            GENERATE_STRUCT_FOR_NULLS_DOC)
+        .define(GENERATE_INDEX_FOR_UNIONS_CONFIG,
+            ConfigDef.Type.BOOLEAN,
+            GENERATE_INDEX_FOR_UNIONS_DEFAULT,
+            ConfigDef.Importance.LOW,
+            GENERATE_INDEX_FOR_UNIONS_DOC
         );
   }
 
@@ -142,6 +152,10 @@ public boolean generateStructForNulls() {
     return this.getBoolean(GENERATE_STRUCT_FOR_NULLS_CONFIG);
   }
 
+  public boolean generateIndexForUnions() {
+    return this.getBoolean(GENERATE_INDEX_FOR_UNIONS_CONFIG);
+  }
+
   public static class Builder {
 
     private final Map<String, Object> props = new HashMap<>();

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

@@ -261,6 +261,45 @@ private SchemaBuilder getEnumUnionSchemaBuilder() {
     return enumUnionBuilder;
   }
 
+  private SchemaBuilder getEnumUnionSchemaBuilderWithoutIndex() {
+    final SchemaBuilder enumUnionBuilder = SchemaBuilder.struct();
+    enumUnionBuilder.name("EnumUnion");
+    final SchemaBuilder someValBuilder = SchemaBuilder.struct();
+    someValBuilder.name("io.confluent.connect.protobuf.Union.some_val");
+    someValBuilder.field(
+        "one_id",
+        SchemaBuilder.string().optional().parameter(PROTOBUF_TYPE_TAG, String.valueOf(1)).build()
+    );
+    someValBuilder.field(
+        "other_id",
+        SchemaBuilder.int32().optional().parameter(PROTOBUF_TYPE_TAG, String.valueOf(2)).build()
+    );
+    someValBuilder.field(
+        "some_status",
+        SchemaBuilder.string()
+            .name("Status")
+            .optional()
+            .parameter(PROTOBUF_TYPE_TAG, String.valueOf(3))
+            .parameter(PROTOBUF_TYPE_ENUM, "Status")
+            .parameter(PROTOBUF_TYPE_ENUM + ".ACTIVE", "0")
+            .parameter(PROTOBUF_TYPE_ENUM + ".INACTIVE", "1")
+            .build()
+    );
+    enumUnionBuilder.field("some_val", someValBuilder.optional().build());
+    enumUnionBuilder.field(
+        "status",
+        SchemaBuilder.string()
+            .name("Status")
+            .optional()
+            .parameter(PROTOBUF_TYPE_TAG, String.valueOf(4))
+            .parameter(PROTOBUF_TYPE_ENUM, "Status")
+            .parameter(PROTOBUF_TYPE_ENUM + ".ACTIVE", "0")
+            .parameter(PROTOBUF_TYPE_ENUM + ".INACTIVE", "1")
+            .build()
+    );
+    return enumUnionBuilder;
+  }
+
   private SchemaBuilder getEnumUnionSchemaBuilderWithGeneralizedSumTypeSupport() {
     final SchemaBuilder enumUnionBuilder = SchemaBuilder.struct();
     enumUnionBuilder.name("EnumUnion");
@@ -311,6 +350,16 @@ private Struct getEnumUnionWithString() throws ParseException {
     return result;
   }
 
+  private Struct getEnumUnionWithStringWithoutIndex() throws ParseException {
+    Schema schema = getEnumUnionSchemaBuilderWithoutIndex().build();
+    Struct result = new Struct(schema.schema());
+    Struct union = new Struct(schema.field("some_val").schema());
+    union.put("one_id", "ID");
+    result.put("some_val", union);
+    result.put("status", "INACTIVE");
+    return result;
+  }
+
   private Struct getEnumUnionWithStringWithGeneralizedSumTypeSupport() throws ParseException {
     Schema schema = getEnumUnionSchemaBuilderWithGeneralizedSumTypeSupport().build();
     Struct result = new Struct(schema.schema());
@@ -702,11 +751,25 @@ public void testToConnectDataDefaultOneOfCannotHaveTwoOneOfsSet() throws Excepti
   public void testToConnectEnumUnionWithString() throws Exception {
     EnumUnion message = createEnumUnionWithString();
     SchemaAndValue result = getSchemaAndValue(message);
-      Schema expectedSchema = getEnumUnionSchemaBuilder().build();
-      assertSchemasEqual(expectedSchema, result.schema());
-      Struct expected = getEnumUnionWithString();
-      assertEquals(expected, result.value());
-    }
+    Schema expectedSchema = getEnumUnionSchemaBuilder().build();
+    assertSchemasEqual(expectedSchema, result.schema());
+    Struct expected = getEnumUnionWithString();
+    assertEquals(expected, result.value());
+  }
+
+  @Test
+  public void testToConnectEnumUnionWithStringWithoutIndex() throws Exception {
+    EnumUnion message = createEnumUnionWithString();
+    ProtobufDataConfig protobufDataConfig = new ProtobufDataConfig.Builder()
+        .with(ProtobufDataConfig.GENERATE_INDEX_FOR_UNIONS_CONFIG, "false")
+        .build();
+    ProtobufData protobufData = new ProtobufData(protobufDataConfig);
+    SchemaAndValue result = getSchemaAndValue(protobufData, message);
+    Schema expectedSchema = getEnumUnionSchemaBuilderWithoutIndex().build();
+    assertSchemasEqual(expectedSchema, result.schema());
+    Struct expected = getEnumUnionWithStringWithoutIndex();
+    assertEquals(expected, result.value());
+  }
 
   @Test
   public void testToConnectEnumUnionWithStringWithGeneralizedSumTypeSupport() throws Exception {