Fixed #508: Issue with Neo4j kafka sink connector in combination with Avro schema nesting (#511)

Author: mroiter-larus

kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/converters/Neo4jValueConverter.kt

@@ -1,5 +1,6 @@
 package streams.kafka.connect.sink.converters
 
+import org.apache.kafka.connect.data.Struct
 import org.neo4j.driver.Value
 import org.neo4j.driver.Values
 import java.math.BigDecimal
@@ -52,4 +53,11 @@ class Neo4jValueConverter: MapValueConverter<Value>() {
         val localDate = value.toInstant().atZone(UTC).toLocalDate()
         setValue(result, fieldName, localDate)
     }
+
+    override fun setStructField(result: MutableMap<String, Value?>?, fieldName: String, value: Struct) {
+        val converted = convert(value)
+            .mapValues { it.value?.asObject() }
+            .toMutableMap() as MutableMap<Any?, Any?>
+        setMap(result, fieldName, null, converted)
+    }
 }

kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jValueConverterTest.kt

@@ -181,6 +181,25 @@ class Neo4jValueConverterTest {
         assertEquals(date.toInstant().atZone(ZoneId.of("UTC")).toLocalDateTime(), result["date"]?.asLocalDateTime())
     }
 
+    @Test
+    fun `should be able to process a nested AVRO structure`() {
+        val trainSchema = SchemaBuilder.struct()
+            .field("internationalTrainNumber", Schema.STRING_SCHEMA)
+            .field("trainDate", Schema.STRING_SCHEMA).build()
+        val mySchema = SchemaBuilder.struct()
+            .field("trainId", trainSchema)
+            .field("coreId", Schema.STRING_SCHEMA).build()
+
+        val trainIdStruct = Struct(trainSchema)
+            .put("internationalTrainNumber", "46261")
+            .put("trainDate", "2021-05-20")
+        val rootStruct = Struct(mySchema)
+            .put("trainId", trainIdStruct)
+            .put("coreId", "000000046261")
+
+        val result = Neo4jValueConverter().convert(rootStruct) as Map<*, *>
+    }
+
     companion object {
         private val LI_SCHEMA = SchemaBuilder.struct().name("org.neo4j.example.html.LI")
                 .field("value", Schema.OPTIONAL_STRING_SCHEMA)