fixes #107 and adds support for multi-level nesting of arrays (#114)

Author: hpgrahsl

src/main/java/at/grahsl/kafka/connect/mongodb/converter/AvroJsonSchemafulRecordConverter.java

@@ -28,6 +28,7 @@
 import org.bson.BsonArray;
 import org.bson.BsonDocument;
 import org.bson.BsonNull;
+import org.bson.BsonValue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -119,7 +120,7 @@ private void processField(BsonDocument doc, Struct struct, Field field) {
                     handleStructField(doc, (Struct)struct.get(field), field);
                     break;
                 case ARRAY:
-                    handleArrayField(doc, (List)struct.get(field), field);
+                    doc.put(field.name(),handleArrayField((List)struct.get(field), field));
                     break;
                 case MAP:
                     handleMapField(doc, (Map)struct.get(field), field);
@@ -157,31 +158,34 @@ private void handleMapField(BsonDocument doc, Map m, Field field) {
                 final List list = (List)m.get(key);
                 logger.trace("adding array values to {} of type valueSchema={} value='{}'",
                    elementField.name(), elementField.schema().valueSchema(), list);
-                handleArrayField(bd, list, elementField);
+                bd.put(key, handleArrayField(list, elementField));
             } else {
                 bd.put(key, toBsonDoc(field.schema().valueSchema(), m.get(key)));
             }
         }
         doc.put(field.name(), bd);
     }
 
-    private void handleArrayField(BsonDocument doc, List list, Field field) {
+    private BsonValue handleArrayField(List list, Field field) {
         logger.trace("handling complex type 'array' of types '{}'",
            field.schema().valueSchema().type());
         if(list==null) {
             logger.trace("no array -> adding null");
-            doc.put(field.name(), BsonNull.VALUE);
-            return;
+            return BsonNull.VALUE;
         }
         BsonArray array = new BsonArray();
+        Schema.Type st = field.schema().valueSchema().type();
         for(Object element : list) {
-            if(field.schema().valueSchema().type().isPrimitive()) {
+            if(st.isPrimitive()) {
                 array.add(getConverter(field.schema().valueSchema()).toBson(element,field.schema()));
+            } else if(st == Schema.Type.ARRAY) {
+                Field elementField = new Field("first", 0, field.schema().valueSchema());
+                array.add(handleArrayField((List)element,elementField));
             } else {
                 array.add(toBsonDoc(field.schema().valueSchema(), element));
             }
         }
-        doc.put(field.name(), array);
+        return array;
     }
 
     private void handleStructField(BsonDocument doc, Struct struct, Field field) {

src/test/java/at/grahsl/kafka/connect/mongodb/converter/RecordConverterTest.java

@@ -61,6 +61,7 @@ public static void initializeTestData() {
                 "\"mySubDoc1\":{\"myString\":\"hello json\"}," +
                 "\"myArray1\":[\"str_1\",\"str_2\",\"...\",\"str_N\"]," +
                 "\"myArray2\":[{\"k\":\"a\",\"v\":1},{\"k\":\"b\",\"v\":2},{\"k\":\"c\",\"v\":3}]," +
+                "\"myArray3\":[[[1],[],[2,3],[4,5,6]]]," +
                 "\"mySubDoc2\":{\"k1\":9,\"k2\":8,\"k3\":7}," +
                 "\"myMapOfStrings\":{\"k1\": [ \"v1-a\", \"v1-b\" ],\"k2\": [ \"v2-a\" ],\"k3\":[ \"v3-a\", \"v3-b\", \"v3-c\" ]}," +
                 "\"myMapOfInts\":{\"k1\": [ 11, 12 ],\"k2\": [ 21 ],\"k3\":[ 31, 32, 33 ]}," +
@@ -85,6 +86,7 @@ public static void initializeTestData() {
                                     .field("v",Schema.INT32_SCHEMA)
                                     .build())
                 )
+                .field("myArray3", SchemaBuilder.array(SchemaBuilder.array(SchemaBuilder.array(Schema.INT32_SCHEMA))))
                 .field("mySubDoc2", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build())
                 .field( "myMapOfStrings", SchemaBuilder.map(Schema.STRING_SCHEMA, SchemaBuilder.array(Schema.STRING_SCHEMA).build()).build())
                 .field( "myMapOfInts", SchemaBuilder.map(Schema.STRING_SCHEMA, SchemaBuilder.array(Schema.INT32_SCHEMA).build()).build())
@@ -113,6 +115,9 @@ public static void initializeTestData() {
                                 .put("k","c").put("v",3)
                         )
                 )
+                .put("myArray3", Arrays.asList(
+                        Arrays.asList(Arrays.asList(1),Arrays.asList(),Arrays.asList(2,3),Arrays.asList(4,5,6))
+                ))
                 .put("mySubDoc2",new HashMap<String,Integer>(){{ put("k1",9); put("k2",8); put("k3",7);}})
                 .put("myMapOfStrings", new HashMap<String, List<String>>(){{
                    put("k1", Arrays.asList("v1-a", "v1-b"));
@@ -152,6 +157,10 @@ public static void initializeTestData() {
                 new HashMap<Object,Object>(){{put("k","c");put("v",3);}}
                 )
         );
+        OBJ_MAP_1.put("myArray3",Arrays.asList(
+                Arrays.asList(Arrays.asList(1), Arrays.asList(), Arrays.asList(2,3), Arrays.asList(4,5,6))
+                )
+        );
         OBJ_MAP_1.put("mySubDoc2",new HashMap<String,Integer>(){{ put("k1",9); put("k2",8); put("k3",7);}});
         OBJ_MAP_1.put("myMapOfStrings",new HashMap<String,List<String>>(){{
                 put("k1",Arrays.asList("v1-a", "v1-b"));
@@ -195,6 +204,14 @@ public static void initializeTestData() {
                         new BsonDocument("k", new BsonString("b")).append("v", new BsonInt32(2)),
                         new BsonDocument("k", new BsonString("c")).append("v", new BsonInt32(3))))
                 )
+                .append("myArray3", new BsonArray(Arrays.asList(
+                        new BsonArray(Arrays.asList(
+                                new BsonArray(Arrays.asList(new BsonInt32(1))),
+                                new BsonArray(),
+                                new BsonArray(Arrays.asList(new BsonInt32(2),new BsonInt32(3))),
+                                new BsonArray(Arrays.asList(new BsonInt32(4),new BsonInt32(5),new BsonInt32(6)))
+                        ))))
+                )
                 .append("mySubDoc2", new BsonDocument("k1", new BsonInt32(9))
                         .append("k2", new BsonInt32(8))
                         .append("k3", new BsonInt32(7))