Fixes #470: Relationships are not generated using CUD file format as sink ingestion strategy (#471)

vga91 · web-flow · commit 3b19274a7785 · 2021-06-07T19:55:51.000+02:00
diff --git a/common/src/main/kotlin/streams/service/sink/strategy/CUDIngestionStrategy.kt b/common/src/main/kotlin/streams/service/sink/strategy/CUDIngestionStrategy.kt
@@ -233,7 +233,7 @@ class CUDIngestionStrategy: IngestionStrategy {
                         try {
                             val data = toCUDEntity<CUDRelationship>(it)
                             when {
-                                data!!.isValidOperation()  -> if (data.from.ids.isNotEmpty() && data.to.ids.isNotEmpty() && data.properties.isNotEmpty()) data else null // TODO send to the DLQ the null
+                                data!!.isValidOperation()  -> if (data.from.ids.isNotEmpty() && data.to.ids.isNotEmpty()) data else null // TODO send to the DLQ the null
                                 else -> null // TODO send to the DLQ the null
                             }
                         } catch (e: Exception) {
@@ -264,7 +264,7 @@ class CUDIngestionStrategy: IngestionStrategy {
                         try {
                             val data = toCUDEntity<CUDRelationship>(it)
                             when (data?.op)  {
-                                CUDOperations.delete -> if (data.from.ids.isNotEmpty() && data.to.ids.isNotEmpty() && data.properties.isEmpty()) data else null // TODO send to the DLQ the null
+                                CUDOperations.delete -> if (data.from.ids.isNotEmpty() && data.to.ids.isNotEmpty()) data else null // TODO send to the DLQ the null
                                 else -> null // TODO send to the DLQ the null
                             }
                         } catch (e: Exception) {
diff --git a/common/src/test/kotlin/streams/service/sink/strategy/CUDIngestionStrategyTest.kt b/common/src/test/kotlin/streams/service/sink/strategy/CUDIngestionStrategyTest.kt
@@ -512,6 +512,54 @@ class CUDIngestionStrategyTest {
         assertRelationshipEventsContainsKey(updateRelFooBarLabel, key, key)
     }
 
+    @Test
+    fun `should create and delete relationship also without properties field`() {
+        val key = "key"
+        val startNode = "SourceNode"
+        val endNode = "TargetNode"
+        val relType = "MY_REL"
+        val start = CUDNodeRel(ids = mapOf(key to 1), labels = listOf(startNode))
+        val end = CUDNodeRel(ids = mapOf(key to 2), labels = listOf(endNode))
+        val list = listOf(CUDOperations.create, CUDOperations.delete, CUDOperations.update).map {
+            val rel = CUDRelationship(op = it, from = start, to = end, rel_type = relType)
+            StreamsSinkEntity(null, rel)
+        }
+
+        // when
+        val cudQueryStrategy = CUDIngestionStrategy()
+        val relationshipEvents = cudQueryStrategy.mergeRelationshipEvents(list)
+        val relationshipDeleteEvents = cudQueryStrategy.deleteRelationshipEvents(list)
+
+        assertEquals(2, relationshipEvents.size)
+        val createRel = findEventByQuery("""
+                |${StreamsUtils.UNWIND}
+                |MATCH (from:$startNode {key: event.from.${CUDIngestionStrategy.ID_KEY}.key})
+                |${StreamsUtils.WITH_EVENT_FROM}
+                |MATCH (to:$endNode {key: event.to.${CUDIngestionStrategy.ID_KEY}.key})
+                |CREATE (from)-[r:$relType]->(to)
+                |SET r = event.properties
+            """.trimMargin(), relationshipEvents)
+        assertEquals(1, createRel.events.size)
+        val updateRel = findEventByQuery("""
+                |${StreamsUtils.UNWIND}
+                |MATCH (from:$startNode {key: event.from.${CUDIngestionStrategy.ID_KEY}.key})
+                |MATCH (to:$endNode {key: event.to.${CUDIngestionStrategy.ID_KEY}.key})
+                |MATCH (from)-[r:MY_REL]->(to)
+                |SET r += event.properties
+            """.trimMargin(), relationshipEvents)
+        assertEquals(1, updateRel.events.size)
+
+        assertEquals(1, relationshipDeleteEvents.size)
+        val deleteRel = findEventByQuery("""
+                |${StreamsUtils.UNWIND}
+                |MATCH (from:$startNode {key: event.from.${CUDIngestionStrategy.ID_KEY}.key})
+                |MATCH (to:$endNode {key: event.to.${CUDIngestionStrategy.ID_KEY}.key})
+                |MATCH (from)-[r:$relType]->(to)
+                |DELETE r
+            """.trimMargin(), relationshipDeleteEvents)
+        assertEquals(1, deleteRel.events.size)
+    }
+
     @Test
     fun `should create, merge and update relationships with merge op in 'from' and 'to' node`() {
         // given
diff --git a/consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkCUDFormatTSE.kt b/consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkCUDFormatTSE.kt
@@ -240,7 +240,7 @@ class KafkaEventSinkCUDFormatTSE : KafkaEventSinkBaseTSE() {
 
         // when
         list.forEach {
-            var producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), it)
+            val producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), it)
             kafkaProducer.send(producerRecord).get()
         }
 
@@ -266,6 +266,57 @@ class KafkaEventSinkCUDFormatTSE : KafkaEventSinkBaseTSE() {
         }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
     }
 
+    @Test
+    fun `should ingest relationship data from CUD Events without properties field`() {
+
+        val relType = "MY_REL"
+        val key = "key"
+        val startNode = "SourceNode"
+        val endNode = "TargetNode"
+        
+        val start = CUDNodeRel(ids = mapOf(key to 1), labels = listOf(startNode))
+        val end = CUDNodeRel(ids = mapOf(key to 1), labels = listOf(endNode))
+        val relMerge = CUDRelationship(op = CUDOperations.merge, from = start, to = end, rel_type = relType)
+        val relMergeAsBytes = JSONUtils.writeValueAsBytes(relMerge)
+
+        val topic = UUID.randomUUID().toString()
+        db.setConfig("streams.sink.topic.cud", topic)
+        db.start()
+        db.beginTx().use {
+            db.execute("""
+                CREATE (:$startNode {key: 1})
+                CREATE (:$endNode {key: 1})
+            """.trimIndent())
+            assertEquals(2, it.allNodes.count())
+            it.commit()
+        }
+
+        val producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), relMergeAsBytes)
+        kafkaProducer.send(producerRecord).get()
+
+        val query = "MATCH p = (:$startNode)-[:$relType]->(:$endNode) RETURN count(p) AS count"
+        
+        Assert.assertEventually(ThrowingSupplier {
+            db.execute(query) {
+                val count = it.columnAs<Long>("count")
+                count.hasNext() && count.next() == 1L && !count.hasNext()
+            }
+        }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
+        
+        val relDelete = CUDRelationship(op = CUDOperations.delete, from = start, to = end, rel_type = relType)
+        val relDeleteAsBytes = JSONUtils.writeValueAsBytes(relDelete)
+
+        val producerRecordDelete = ProducerRecord(topic, UUID.randomUUID().toString(), relDeleteAsBytes)
+        kafkaProducer.send(producerRecordDelete).get()
+        
+        Assert.assertEventually(ThrowingSupplier {
+            db.execute(query) {
+                val count = it.columnAs<Long>("count")
+                count.hasNext() && count.next() == 0L && !count.hasNext()
+            }
+        }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
+    }
+
     @Test
     fun `should delete relationship data from CUD Events`() {
         // given
diff --git a/kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkTaskTest.kt b/kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkTaskTest.kt
@@ -987,6 +987,56 @@ class Neo4jSinkTaskTest {
         }
     }
 
+    @Test
+    fun `should create and delete relationship from CUD event without properties field`() {
+        val relType = "MY_REL"
+        val key = "key"
+        val startNode = "SourceNode"
+        val endNode = "TargetNode"
+        val topic = UUID.randomUUID().toString()
+
+        db.defaultDatabaseService().beginTx().use {
+            it.execute("CREATE (:$startNode {key: 1}) CREATE (:$endNode {key: 1})").close()
+            it.commit()
+        }
+        
+        val start = CUDNodeRel(ids = mapOf(key to 1), labels = listOf(startNode))
+        val end = CUDNodeRel(ids = mapOf(key to 1), labels = listOf(endNode))
+        val relMerge = CUDRelationship(op = CUDOperations.merge, from = start, to = end, rel_type = relType)
+        val sinkRecordMerge = SinkRecord(topic, 1, null, null, null, JSONUtils.asMap(relMerge), 0L)
+
+        val props = mutableMapOf<String, String>()
+        props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
+        props[Neo4jSinkConnectorConfig.TOPIC_CUD] = topic
+        props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
+        props[SinkTask.TOPICS_CONFIG] = topic
+
+        val task = Neo4jSinkTask()
+        task.initialize(mock(SinkTaskContext::class.java))
+        task.start(props)
+        task.put(listOf(sinkRecordMerge))
+
+        val queryCount = "MATCH p = (:$startNode)-[:$relType]->(:$endNode) RETURN count(p) AS count"
+        
+        db.defaultDatabaseService().beginTx().use {
+            val countRels = it.execute(queryCount)
+                .columnAs<Long>("count")
+                .next()
+            assertEquals(1L, countRels)
+        }
+
+        val relDelete = CUDRelationship(op = CUDOperations.delete, from = start, to = end, rel_type = relType)
+        val sinkRecordDelete = SinkRecord(topic, 1, null, null, null, JSONUtils.asMap(relDelete), 1L)
+        task.put(listOf(sinkRecordDelete))
+
+        db.defaultDatabaseService().beginTx().use {
+            val countRels = it.execute(queryCount)
+                .columnAs<Long>("count")
+                .next()
+            assertEquals(0L, countRels)
+        }
+    }
+
     @Test
     fun `should ingest node data from CUD Events`() {
         // given
