Support dotted field names in avro

KAFKA-137

Author: rozza

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

@@ -113,7 +113,7 @@ public static Schema fromJson(final String jsonSchema) {
 
   static org.apache.avro.Schema parseSchema(final String jsonSchema) {
     try {
-      return new Parser().parse(jsonSchema);
+      return new Parser().setValidate(false).parse(jsonSchema);
     } catch (Exception e) {
       throw new ConnectException(format("Invalid Avro schema. %s\n%s", e.getMessage(), jsonSchema));
     }

src/test/java/com/mongodb/kafka/connect/source/schema/AvroSchemaTest.java

@@ -198,6 +198,27 @@ void testUnsupportedAvroSchema() {
         });
   }
 
+  @Test
+  @DisplayName("test nested field support")
+  void testNestedFieldSupport() {
+    Schema actual =
+        AvroSchema.fromJson(
+            "{\n"
+                + "  \"type\": \"record\",\n"
+                + "  \"name\": \"keySchema\",\n"
+                + "  \"fields\" : [{\"name\": \"fullDocument.documentKey\", \"type\": \"string\", \"default\": \"MISSING\"}]\n"
+                + "}");
+
+    Schema expected =
+        SchemaBuilder.struct()
+            .name("keySchema")
+            .field(
+                "fullDocument.documentKey", SchemaBuilder.string().defaultValue("MISSING").build())
+            .build();
+
+    SchemaUtils.assertSchemaEquals(expected, actual);
+  }
+
   @Test
   @DisplayName("test invalid avro schema definitions")
   void testInvalidAvroSchema() {