Fix Mongo inferred names for schema records

Author: rajdangwal

src/main/java/com/mongodb/kafka/connect/source/producer/InferSchemaAndValueProducer.java

@@ -35,6 +35,6 @@ final class InferSchemaAndValueProducer implements SchemaAndValueProducer {
   @Override
   public SchemaAndValue get(final BsonDocument changeStreamDocument) {
     return bsonValueToSchemaAndValue.toSchemaAndValue(
-        inferDocumentSchema(changeStreamDocument, false), changeStreamDocument);
+        inferDocumentSchema(changeStreamDocument, false, "default"), changeStreamDocument);
   }
 }

src/main/java/com/mongodb/kafka/connect/source/schema/BsonDocumentToSchema.java

@@ -36,23 +36,24 @@ public final class BsonDocumentToSchema {
   private static final Schema DEFAULT_INFER_SCHEMA_TYPE = Schema.OPTIONAL_STRING_SCHEMA;
   public static final String SCHEMA_NAME_TEMPLATE = "inferred_name_%s";
 
-  public static Schema inferDocumentSchema(final BsonDocument document, final boolean optional) {
+  public static Schema inferDocumentSchema(
+      final BsonDocument document, final boolean optional, final String fieldName) {
     SchemaBuilder builder = SchemaBuilder.struct();
     if (document.containsKey(ID_FIELD)) {
-      builder.field(ID_FIELD, inferSchema(document.get(ID_FIELD)));
+      builder.field(ID_FIELD, inferSchema(document.get(ID_FIELD), ID_FIELD));
     }
     document.entrySet().stream()
         .filter(kv -> !kv.getKey().equals(ID_FIELD))
         .sorted(Map.Entry.comparingByKey())
-        .forEach(kv -> builder.field(kv.getKey(), inferSchema(kv.getValue())));
-    builder.name(generateName(builder));
+        .forEach(kv -> builder.field(kv.getKey(), inferSchema(kv.getValue(), kv.getKey())));
+    builder.name(generateName(builder, fieldName));
     if (optional) {
       builder.optional();
     }
     return builder.build();
   }
 
-  public static Schema inferSchema(final BsonValue bsonValue) {
+  public static Schema inferSchema(final BsonValue bsonValue, final String fieldName) {
     switch (bsonValue.getBsonType()) {
       case BOOLEAN:
         return Schema.OPTIONAL_BOOLEAN_SCHEMA;
@@ -70,16 +71,17 @@ public static Schema inferSchema(final BsonValue bsonValue) {
       case TIMESTAMP:
         return Timestamp.builder().optional().build();
       case DOCUMENT:
-        return inferDocumentSchema(bsonValue.asDocument(), true);
+        return inferDocumentSchema(bsonValue.asDocument(), true, fieldName);
       case ARRAY:
         List<BsonValue> values = bsonValue.asArray().getValues();
         Schema firstItemSchema =
-            values.isEmpty() ? DEFAULT_INFER_SCHEMA_TYPE : inferSchema(values.get(0));
+            values.isEmpty() ? DEFAULT_INFER_SCHEMA_TYPE : inferSchema(values.get(0), fieldName);
         if (values.isEmpty()
-            || values.stream().anyMatch(bv -> !Objects.equals(inferSchema(bv), firstItemSchema))) {
+            || values.stream()
+                .anyMatch(bv -> !Objects.equals(inferSchema(bv, fieldName), firstItemSchema))) {
           return SchemaBuilder.array(DEFAULT_INFER_SCHEMA_TYPE).optional().build();
         }
-        return SchemaBuilder.array(inferSchema(bsonValue.asArray().getValues().get(0)))
+        return SchemaBuilder.array(inferSchema(bsonValue.asArray().getValues().get(0), fieldName))
             .optional()
             .build();
       case BINARY:
@@ -101,8 +103,9 @@ public static Schema inferSchema(final BsonValue bsonValue) {
     }
   }
 
-  public static String generateName(final SchemaBuilder builder) {
-    return format(SCHEMA_NAME_TEMPLATE, Objects.hashCode(builder.build())).replace("-", "_");
+  public static String generateName(final SchemaBuilder builder, final String fieldName) {
+    builder.build();
+    return format(SCHEMA_NAME_TEMPLATE, fieldName);
   }
 
   private BsonDocumentToSchema() {}

src/test/java/com/mongodb/kafka/connect/source/producer/SchemaAndValueProducerTest.java

@@ -156,14 +156,16 @@ void testAvroSchemaAndValueProducer() {
   @DisplayName("test infer schema and value producer")
   void testInferSchemaAndValueProducer() {
 
+    String fieldName = "default";
     Schema expectedSchema =
         nameAndBuildSchema(
             SchemaBuilder.struct()
                 .field(
                     "arrayComplex",
                     SchemaBuilder.array(
                             nameAndBuildOptionalSchema(
-                                SchemaBuilder.struct().field("a", Schema.OPTIONAL_INT32_SCHEMA)))
+                                SchemaBuilder.struct().field("a", Schema.OPTIONAL_INT32_SCHEMA),
+                                "a"))
                         .optional()
                         .build())
                 .field(
@@ -187,7 +189,8 @@ void testInferSchemaAndValueProducer() {
                 .field(
                     "document",
                     nameAndBuildOptionalSchema(
-                        SchemaBuilder.struct().field("a", Schema.OPTIONAL_INT32_SCHEMA)))
+                        SchemaBuilder.struct().field("a", Schema.OPTIONAL_INT32_SCHEMA),
+                        "document"))
                 .field("double", Schema.OPTIONAL_FLOAT64_SCHEMA)
                 .field("int32", Schema.OPTIONAL_INT32_SCHEMA)
                 .field("int64", Schema.OPTIONAL_INT64_SCHEMA)
@@ -199,7 +202,8 @@ void testInferSchemaAndValueProducer() {
                 .field("string", Schema.OPTIONAL_STRING_SCHEMA)
                 .field("symbol", Schema.OPTIONAL_STRING_SCHEMA)
                 .field("timestamp", Timestamp.builder().optional().build())
-                .field("undefined", Schema.OPTIONAL_STRING_SCHEMA));
+                .field("undefined", Schema.OPTIONAL_STRING_SCHEMA),
+            fieldName);
 
     Schema arrayComplexValueSchema = expectedSchema.field("arrayComplex").schema().valueSchema();
     Schema documentSchema = expectedSchema.field("document").schema();
@@ -336,12 +340,12 @@ static Struct generateExpectedValue(final boolean simplified) {
     };
   }
 
-  static Schema nameAndBuildSchema(final SchemaBuilder builder) {
-    return builder.name(generateName(builder)).build();
+  static Schema nameAndBuildSchema(final SchemaBuilder builder, final String fieldName) {
+    return builder.name(generateName(builder, fieldName)).build();
   }
 
-  static Schema nameAndBuildOptionalSchema(final SchemaBuilder builder) {
-    return builder.name(generateName(builder)).optional().build();
+  static Schema nameAndBuildOptionalSchema(final SchemaBuilder builder, final String fieldName) {
+    return builder.name(generateName(builder, fieldName)).optional().build();
   }
 
   static String getFullDocument(final boolean simplified) {