Merge branch '7.2.x' into 7.3.x

confluentinc · Sep 29, 2022 · 9d7e8d3 · 9d7e8d3
2 parents 0b1f1fa + 384af90
commit 9d7e8d3
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 6 deletions.
diff --git a/protobuf-converter/src/main/java/io/confluent/connect/protobuf/ProtobufData.java b/protobuf-converter/src/main/java/io/confluent/connect/protobuf/ProtobufData.java
@@ -1667,7 +1667,8 @@ private static boolean isMapDescriptor(
 
   private SchemaBuilder toMapSchema(ToConnectContext ctx, Descriptor descriptor) {
     List<FieldDescriptor> fieldDescriptors = descriptor.getFields();
-    String name = ProtobufSchema.toMapField(descriptor.getName());
+    String name = ProtobufSchema.toMapField(
+        enhancedSchemaSupport ? descriptor.getFullName() : descriptor.getName());
     return SchemaBuilder.map(toConnectSchema(ctx, fieldDescriptors.get(0)),
         toConnectSchema(ctx, fieldDescriptors.get(1))
     ).name(name);

diff --git a/protobuf-converter/src/test/java/io/confluent/connect/protobuf/ProtobufDataTest.java b/protobuf-converter/src/test/java/io/confluent/connect/protobuf/ProtobufDataTest.java
@@ -2178,6 +2178,36 @@ public void testToConnectFullyQualifiedSchema() {
     assertEquals("foo.Event.Action", actual.field("action").schema().name());
   }
 
+  @Test
+  public void testToConnectFullyQualifiedMapSchema() {
+    String schema = "syntax = \"proto3\";\n"
+        + "\n"
+        + "option java_package = \"io.confluent.connect.protobuf.test\";\n"
+        + "\n"
+        + "message Customer {\n"
+        + "  map<string,string> tags = 1;\n"
+        + "  Meta meta = 2;\n"
+        + "}\n"
+        + "\n"
+        + "message Meta {\n"
+        + "  map<string,Value> tags = 2;\n"
+        + "}\n"
+        + "\n"
+        + "message Value{\n"
+        + "  float a=1;\n"
+        + "  float b=2;\n"
+        + "}\n";
+
+    ProtobufSchema protobufSchema = new ProtobufSchema(schema);
+    Map<String, Object> configs = new HashMap<>();
+    configs.put(ProtobufDataConfig.ENHANCED_PROTOBUF_SCHEMA_SUPPORT_CONFIG, true);
+    ProtobufData protobufData = new ProtobufData(new ProtobufDataConfig(configs));
+    Schema actual = protobufData.toConnectSchema(protobufSchema);
+    assertEquals("Customer.tags",
+        actual.field("tags").schema().name());
+    assertEquals("Meta.tags", actual.field("meta").schema().field("tags").schema().name());
+  }
+
   @Test
   public void testToConnectMultipleMapReferences() throws Exception {
     AttributeFieldEntry entry1 = AttributeFieldEntry.newBuilder()

diff --git a/...buf-provider/src/main/java/io/confluent/kafka/schemaregistry/protobuf/ProtobufSchema.java b/...buf-provider/src/main/java/io/confluent/kafka/schemaregistry/protobuf/ProtobufSchema.java
@@ -1919,11 +1919,14 @@ public static String toMapEntry(String s) {
   }
 
   public static String toMapField(String s) {
-    if (s.endsWith(MAP_ENTRY_SUFFIX)) {
-      s = s.substring(0, s.length() - MAP_ENTRY_SUFFIX.length());
-      s = UPPER_CAMEL.to(LOWER_UNDERSCORE, s);
-    }
-    return s;
+    String[] parts = s.split("\\.");
+    String lastPart = parts[parts.length - 1];
+    if (lastPart.endsWith(MAP_ENTRY_SUFFIX)) {
+      lastPart = lastPart.substring(0, lastPart.length() - MAP_ENTRY_SUFFIX.length());
+      lastPart = UPPER_CAMEL.to(LOWER_UNDERSCORE, lastPart);
+      parts[parts.length - 1] = lastPart;
+    }
+    return String.join(".", parts);
   }
 
   public enum Format {