Issue #542: Add a strategy to merge edge nodes properties when creating relationships using Relationship pattern strategy (#544)

* Issue #542: Add a strategy to merge edge nodes properties when creating relationships using Relationship pattern strategy

* some refactoring

Co-authored-by: Andrea Santurbano <[email protected]>

Author: mroiter-larus

.gitignore

@@ -31,4 +31,3 @@ Thumbs.db
 bin
 doc/node
 doc/node_modules
-

common/src/main/kotlin/streams/service/sink/strategy/NodePatternIngestionStrategy.kt

@@ -15,7 +15,7 @@ class NodePatternIngestionStrategy(private val nodePatternConfiguration: NodePat
                 |MERGE (n${getLabelsAsString(nodePatternConfiguration.labels)}{${
                     getNodeMergeKeys("keys", nodePatternConfiguration.keys)
                 }})
-                |SET n = event.properties
+                |SET n ${if (nodePatternConfiguration.mergeProperties) "+" else ""}= event.properties
                 |SET n += event.keys
             """.trimMargin()
 

common/src/main/kotlin/streams/service/sink/strategy/PatternConfiguration.kt

@@ -42,14 +42,14 @@ private fun cleanProperties(type: PatternConfigurationType, properties: List<Str
 interface PatternConfiguration
 
 data class NodePatternConfiguration(val keys: Set<String>, val type: PatternConfigurationType,
-                                    val labels: List<String>, val properties: List<String>): PatternConfiguration {
+                                    val labels: List<String>, val properties: List<String>, val mergeProperties: Boolean): PatternConfiguration {
     companion object {
 
         // (:LabelA{!id,foo,bar})
         @JvmStatic private val cypherNodePatternConfigured = """\((:\w+\s*(?::\s*(?:\w+)\s*)*)\s*(?:\{\s*(-?[\w!\.]+\s*(?:,\s*-?[!\w\*\.]+\s*)*)\})?\)$""".toRegex()
         // LabelA{!id,foo,bar}
         @JvmStatic private val simpleNodePatternConfigured = """^(\w+\s*(?::\s*(?:\w+)\s*)*)\s*(?:\{\s*(-?[\w!\.]+\s*(?:,\s*-?[!\w\*\.]+\s*)*)\})?$""".toRegex()
-        fun parse(pattern: String): NodePatternConfiguration {
+        fun parse(pattern: String, mergeProperties: Boolean): NodePatternConfiguration {
             val isCypherPattern = pattern.startsWith("(")
             val regex = if (isCypherPattern) cypherNodePatternConfigured else simpleNodePatternConfigured
             val matcher = regex.matchEntire(pattern)
@@ -75,25 +75,26 @@ data class NodePatternConfiguration(val keys: Set<String>, val type: PatternConf
                 val cleanedProperties = cleanProperties(type, properties)
 
                 return NodePatternConfiguration(keys = keys, type = type,
-                        labels = labels, properties = cleanedProperties)
+                        labels = labels, properties = cleanedProperties, mergeProperties)
             }
         }
     }
+
 }
 
 
 data class RelationshipPatternConfiguration(val start: NodePatternConfiguration, val end: NodePatternConfiguration,
                                             val relType: String, val type: PatternConfigurationType,
-                                            val properties: List<String>): PatternConfiguration {
+                                            val properties: List<String>, val mergeProperties: Boolean): PatternConfiguration {
     companion object {
 
         // we don't allow ALL for start/end nodes in rels
         // it's public for testing purpose
-        fun getNodeConf(pattern: String): NodePatternConfiguration {
-            val start = NodePatternConfiguration.parse(pattern)
+        fun getNodeConf(pattern: String, mergeProperties: Boolean): NodePatternConfiguration {
+            val start = NodePatternConfiguration.parse(pattern, mergeProperties)
             return if (start.type == PatternConfigurationType.ALL) {
                 NodePatternConfiguration(keys = start.keys, type = PatternConfigurationType.INCLUDE,
-                        labels = start.labels, properties = start.properties)
+                        labels = start.labels, properties = start.properties, mergeProperties)
             } else {
                 start
             }
@@ -148,7 +149,7 @@ data class RelationshipPatternConfiguration(val start: NodePatternConfiguration,
             }
         }
 
-        fun parse(pattern: String): RelationshipPatternConfiguration {
+        fun parse(pattern: String, mergeNodeProps: Boolean, mergeRelProps: Boolean): RelationshipPatternConfiguration {
             val isCypherPattern = pattern.startsWith("(")
             val regex = if (isCypherPattern) {
                 cypherRelationshipPatternConfigured
@@ -169,20 +170,20 @@ data class RelationshipPatternConfiguration(val start: NodePatternConfiguration,
                 val metadata = RelationshipPatternMetaData.create(isCypherPattern, isLeftToRight, matcher.groupValues)
 
                 val start = try {
-                    getNodeConf(metadata.startPattern)
+                    getNodeConf(metadata.startPattern, mergeNodeProps)
                 } catch (e: Exception) {
                     throw IllegalArgumentException("The Relationship pattern $pattern is invalid")
                 }
                 val end = try {
-                    getNodeConf(metadata.endPattern)
+                    getNodeConf(metadata.endPattern, mergeNodeProps)
                 } catch (e: Exception) {
                     throw IllegalArgumentException("The Relationship pattern $pattern is invalid")
                 }
                 val type = getPatternConfiguredType(metadata.properties)
                 isHomogeneousPattern(type, metadata.properties, pattern, "Relationship")
                 val cleanedProperties = cleanProperties(type, metadata.properties)
                 return RelationshipPatternConfiguration(start = start, end = end, relType = metadata.relType,
-                        properties = cleanedProperties, type = type)
+                        properties = cleanedProperties, type = type, mergeProperties = mergeRelProps)
             }
         }
 

common/src/main/kotlin/streams/service/sink/strategy/RelationshipPatternIngestionStrategy.kt

@@ -15,15 +15,15 @@ class RelationshipPatternIngestionStrategy(private val relationshipPatternConfig
                 |MERGE (start${getLabelsAsString(relationshipPatternConfiguration.start.labels)}{${
                     getNodeMergeKeys("start.keys", relationshipPatternConfiguration.start.keys)
                 }})
-                |SET start = event.start.properties
+                |SET start ${if (relationshipPatternConfiguration.mergeProperties) "+" else ""}= event.start.properties
                 |SET start += event.start.keys
                 |MERGE (end${getLabelsAsString(relationshipPatternConfiguration.end.labels)}{${
                     getNodeMergeKeys("end.keys", relationshipPatternConfiguration.end.keys)
                 }})
-                |SET end = event.end.properties
+                |SET end ${if (relationshipPatternConfiguration.mergeProperties) "+" else ""}= event.end.properties
                 |SET end += event.end.keys
                 |MERGE (start)-[r:${relationshipPatternConfiguration.relType}]->(end)
-                |SET r = event.properties
+                |SET r ${if (relationshipPatternConfiguration.mergeProperties) "+" else ""}= event.properties
             """.trimMargin()
 
     private val deleteRelationshipTemplate: String = """

common/src/main/kotlin/streams/service/sink/strategy/SourceIdIngestionStrategy.kt

@@ -7,7 +7,11 @@ import streams.utils.IngestionUtils.getLabelsAsString
 import streams.utils.SchemaUtils
 import streams.utils.StreamsUtils
 
-data class SourceIdIngestionStrategyConfig(val labelName: String = "SourceEvent", val idName: String = "sourceId")
+data class SourceIdIngestionStrategyConfig(val labelName: String = "SourceEvent", val idName: String = "sourceId") {
+    companion object {
+        val DEFAULT = SourceIdIngestionStrategyConfig()
+    }
+}
 
 class SourceIdIngestionStrategy(config: SourceIdIngestionStrategyConfig = SourceIdIngestionStrategyConfig()): IngestionStrategy {
 

common/src/test/kotlin/streams/service/sink/strategy/NodePatternIngestionStrategyTest.kt

@@ -10,7 +10,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should get all properties`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id})", true)
         val strategy = NodePatternIngestionStrategy(config)
         val data = mapOf("id" to 1, "foo" to "foo", "bar" to "bar", "foobar" to "foobar")
 
@@ -22,7 +22,7 @@ class NodePatternIngestionStrategyTest {
         assertEquals("""
                 |${StreamsUtils.UNWIND}
                 |MERGE (n:LabelA:LabelB{id: event.keys.id})
-                |SET n = event.properties
+                |SET n += event.properties
                 |SET n += event.keys
             """.trimMargin(), queryEvents[0].query)
         assertEquals(listOf(mapOf("keys" to mapOf("id" to 1),
@@ -37,7 +37,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should get nested properties`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id, foo.bar})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id, foo.bar})", false)
         val strategy = NodePatternIngestionStrategy(config)
         val data = mapOf("id" to 1, "foo" to mapOf("bar" to "bar", "foobar" to "foobar"))
 
@@ -65,7 +65,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should exclude nested properties`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id, -foo})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id, -foo})", false)
         val strategy = NodePatternIngestionStrategy(config)
         val map = mapOf("id" to 1, "foo" to mapOf("bar" to "bar", "foobar" to "foobar"), "prop" to 100)
 
@@ -93,7 +93,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should include nested properties`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id, foo})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id, foo})", true)
         val strategy = NodePatternIngestionStrategy(config)
         val data = mapOf("id" to 1, "foo" to mapOf("bar" to "bar", "foobar" to "foobar"), "prop" to 100)
 
@@ -106,7 +106,7 @@ class NodePatternIngestionStrategyTest {
         assertEquals("""
                 |${StreamsUtils.UNWIND}
                 |MERGE (n:LabelA:LabelB{id: event.keys.id})
-                |SET n = event.properties
+                |SET n += event.properties
                 |SET n += event.keys
             """.trimMargin(),
                 queryEvents[0].query)
@@ -121,7 +121,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should exclude the properties`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id,-foo,-bar})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id,-foo,-bar})", false)
         val strategy = NodePatternIngestionStrategy(config)
         val data = mapOf("id" to 1, "foo" to "foo", "bar" to "bar", "foobar" to "foobar")
 
@@ -146,7 +146,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should include the properties`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id,foo,bar})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id,foo,bar})", false)
         val strategy = NodePatternIngestionStrategy(config)
         val data = mapOf("id" to 1, "foo" to "foo", "bar" to "bar", "foobar" to "foobar")
 
@@ -170,7 +170,7 @@ class NodePatternIngestionStrategyTest {
     @Test
     fun `should delete the node`() {
         // given
-        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id})")
+        val config = NodePatternConfiguration.parse("(:LabelA:LabelB{!id})", true)
         val strategy = NodePatternIngestionStrategy(config)
         val data = mapOf("id" to 1, "foo" to "foo", "bar" to "bar", "foobar" to "foobar")
 

common/src/test/kotlin/streams/service/sink/strategy/PatternConfigurationTest.kt

@@ -13,7 +13,7 @@ class RelationshipPatternIngestionStrategyTest {
         val startPattern = "LabelA{!idStart}"
         val endPattern = "LabelB{!idEnd}"
         val pattern = "(:$startPattern)-[:REL_TYPE]->(:$endPattern)"
-        val config = RelationshipPatternConfiguration.parse(pattern)
+        val config = RelationshipPatternConfiguration.parse(pattern, mergeNodeProps = false, mergeRelProps = true)
         val strategy = RelationshipPatternIngestionStrategy(config)
         val data = mapOf("idStart" to 1, "idEnd" to 2,
                 "foo" to "foo",
@@ -28,13 +28,13 @@ class RelationshipPatternIngestionStrategyTest {
         assertEquals("""
             |${StreamsUtils.UNWIND}
             |MERGE (start:LabelA{idStart: event.start.keys.idStart})
-            |SET start = event.start.properties
+            |SET start += event.start.properties
             |SET start += event.start.keys
             |MERGE (end:LabelB{idEnd: event.end.keys.idEnd})
-            |SET end = event.end.properties
+            |SET end += event.end.properties
             |SET end += event.end.keys
             |MERGE (start)-[r:REL_TYPE]->(end)
-            |SET r = event.properties
+            |SET r += event.properties
         """.trimMargin(), queryEvents[0].query)
         assertEquals(listOf(mapOf("start" to mapOf("keys" to mapOf("idStart" to 1), "properties" to emptyMap()),
                 "end" to mapOf("keys" to mapOf("idEnd" to 2), "properties" to emptyMap()),
@@ -50,7 +50,7 @@ class RelationshipPatternIngestionStrategyTest {
         val startPattern = "LabelA{!idStart}"
         val endPattern = "LabelB{!idEnd}"
         val pattern = "$startPattern REL_TYPE $endPattern"
-        val config = RelationshipPatternConfiguration.parse(pattern)
+        val config = RelationshipPatternConfiguration.parse(pattern, mergeNodeProps = false, mergeRelProps = false)
         val strategy = RelationshipPatternIngestionStrategy(config)
         val data = mapOf("idStart" to 1, "idEnd" to 2,
                 "foo" to "foo",
@@ -87,7 +87,7 @@ class RelationshipPatternIngestionStrategyTest {
         val startPattern = "LabelA{!idStart}"
         val endPattern = "LabelB{!idEnd}"
         val pattern = "(:$endPattern)<-[:REL_TYPE]-(:$startPattern)"
-        val config = RelationshipPatternConfiguration.parse(pattern)
+        val config = RelationshipPatternConfiguration.parse(pattern, mergeNodeProps = false, mergeRelProps = false)
         val strategy = RelationshipPatternIngestionStrategy(config)
         val data = mapOf("idStart" to 1, "idEnd" to 2,
                 "foo" to "foo",
@@ -124,7 +124,7 @@ class RelationshipPatternIngestionStrategyTest {
         val startPattern = "LabelA{!idStart, foo.mapFoo}"
         val endPattern = "LabelB{!idEnd, bar.mapBar}"
         val pattern = "(:$startPattern)-[:REL_TYPE]->(:$endPattern)"
-        val config = RelationshipPatternConfiguration.parse(pattern)
+        val config = RelationshipPatternConfiguration.parse(pattern, mergeNodeProps = false, mergeRelProps = true)
         val strategy = RelationshipPatternIngestionStrategy(config)
         val data = mapOf("idStart" to 1, "idEnd" to 2,
                 "foo" to mapOf("mapFoo" to "mapFoo"),
@@ -141,13 +141,13 @@ class RelationshipPatternIngestionStrategyTest {
         assertEquals("""
             |${StreamsUtils.UNWIND}
             |MERGE (start:LabelA{idStart: event.start.keys.idStart})
-            |SET start = event.start.properties
+            |SET start += event.start.properties
             |SET start += event.start.keys
             |MERGE (end:LabelB{idEnd: event.end.keys.idEnd})
-            |SET end = event.end.properties
+            |SET end += event.end.properties
             |SET end += event.end.keys
             |MERGE (start)-[r:REL_TYPE]->(end)
-            |SET r = event.properties
+            |SET r += event.properties
         """.trimMargin(), queryEvents[0].query)
         assertEquals(listOf(
                 mapOf("start" to mapOf("keys" to mapOf("idStart" to 1), "properties" to mapOf("foo.mapFoo" to "mapFoo")),
@@ -165,7 +165,7 @@ class RelationshipPatternIngestionStrategyTest {
         val startPattern = "LabelA{!idStart}"
         val endPattern = "LabelB{!idEnd}"
         val pattern = "(:$startPattern)-[:REL_TYPE]->(:$endPattern)"
-        val config = RelationshipPatternConfiguration.parse(pattern)
+        val config = RelationshipPatternConfiguration.parse(pattern, mergeNodeProps = false, mergeRelProps = false)
         val strategy = RelationshipPatternIngestionStrategy(config)
         val data = mapOf("idStart" to 1, "idEnd" to 2,
                 "foo" to "foo",

common/src/test/kotlin/streams/service/sink/strategy/RelationshipPatternIngestionStrategyTest.kt

@@ -28,9 +28,8 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): Neo4jConnectorConfig(confi
     val kafkaBrokerProperties: Map<String, Any?>
 
     init {
-        val sourceIdStrategyConfig = SourceIdIngestionStrategyConfig(getString(TOPIC_CDC_SOURCE_ID_LABEL_NAME), getString(TOPIC_CDC_SOURCE_ID_ID_NAME))
         topics = Topics.from(originals as Map<String, Any?>, "streams.sink." to "neo4j.")
-        strategyMap = TopicUtils.toStrategyMap(topics, sourceIdStrategyConfig)
+        strategyMap = TopicUtils.toStrategyMap(topics)
 
         parallelBatches = getBoolean(BATCH_PARALLELIZE)
         val kafkaPrefix = "kafka."
@@ -68,9 +67,18 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): Neo4jConnectorConfig(confi
         const val TOPIC_CDC_SOURCE_ID_ID_NAME = "neo4j.topic.cdc.sourceId.idName"
         const val TOPIC_PATTERN_NODE_PREFIX = "neo4j.topic.pattern.node."
         const val TOPIC_PATTERN_RELATIONSHIP_PREFIX = "neo4j.topic.pattern.relationship."
+        const val TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED = "neo4j.topic.pattern.merge.node.properties.enabled"
+        const val TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED = "neo4j.topic.pattern.merge.relationship.properties.enabled"
         const val TOPIC_CDC_SCHEMA = "neo4j.topic.cdc.schema"
         const val TOPIC_CUD = "neo4j.topic.cud"
 
+
+        const val DEFAULT_BATCH_PARALLELIZE = true
+        const val DEFAULT_TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED = false
+        const val DEFAULT_TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED = false
+
+
+
         private val sourceIdIngestionStrategyConfig = SourceIdIngestionStrategyConfig()
 
         fun config(): ConfigDef = Neo4jConnectorConfig.config()
@@ -92,11 +100,19 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): Neo4jConnectorConfig(confi
                             .build())
                     .define(ConfigKeyBuilder.of(BATCH_PARALLELIZE, ConfigDef.Type.BOOLEAN)
                             .documentation(PropertiesUtil.getProperty(BATCH_PARALLELIZE)).importance(ConfigDef.Importance.MEDIUM)
-                            .defaultValue(true).group(ConfigGroup.BATCH)
+                            .defaultValue(DEFAULT_BATCH_PARALLELIZE).group(ConfigGroup.BATCH)
                             .build())
                     .define(ConfigKeyBuilder.of(TOPIC_CUD, ConfigDef.Type.STRING)
                             .documentation(PropertiesUtil.getProperty(TOPIC_CUD)).importance(ConfigDef.Importance.HIGH)
                             .defaultValue("").group(ConfigGroup.TOPIC_CYPHER_MAPPING)
                             .build())
+                    .define(ConfigKeyBuilder.of(TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED, ConfigDef.Type.BOOLEAN)
+                        .documentation(PropertiesUtil.getProperty(TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED)).importance(ConfigDef.Importance.MEDIUM)
+                        .defaultValue(DEFAULT_TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED).group(ConfigGroup.TOPIC_CYPHER_MAPPING)
+                        .build())
+                    .define(ConfigKeyBuilder.of(TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED, ConfigDef.Type.BOOLEAN)
+                        .documentation(PropertiesUtil.getProperty(TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED)).importance(ConfigDef.Importance.MEDIUM)
+                        .defaultValue(DEFAULT_TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED).group(ConfigGroup.TOPIC_CYPHER_MAPPING)
+                        .build())
     }
 }

doc/package-lock.json

@@ -18,6 +18,7 @@ class Neo4jStrategyStorage(val config: Neo4jSinkConnectorConfig): StreamsStrateg
         when (val topicConfig = topicConfigMap.getOrDefault(topicType, emptyList<Any>())) {
             is Collection<*> -> topicConfig.contains(topic)
             is Map<*, *> -> topicConfig.containsKey(topic)
+            is Pair<*, *> -> (topicConfig.first as Set<String>).contains(topic)
             else -> false
         }
     }

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

@@ -1,5 +1,6 @@
 package streams.service
 
+import streams.kafka.connect.sink.Neo4jSinkConnectorConfig
 import streams.service.sink.strategy.*
 import kotlin.reflect.jvm.javaType
 
@@ -11,29 +12,22 @@ private fun TopicType.replaceKeyBy(replacePrefix: Pair<String, String>) = if (re
         this.key.replace(replacePrefix.first, replacePrefix.second)
 
 data class Topics(val cypherTopics: Map<String, String> = emptyMap(),
-                  val cdcSourceIdTopics: Set<String> = emptySet(),
+                  val cdcSourceIdTopics: Pair<Set<String>, SourceIdIngestionStrategyConfig> = (emptySet<String>() to SourceIdIngestionStrategyConfig()),
                   val cdcSchemaTopics: Set<String> = emptySet(),
                   val cudTopics: Set<String> = emptySet(),
                   val nodePatternTopics: Map<String, NodePatternConfiguration> = emptyMap(),
                   val relPatternTopics: Map<String, RelationshipPatternConfiguration> = emptyMap(),
                   val invalid: List<String> = emptyList()) {
 
-    operator fun plus(other: Topics): Topics {
-        return Topics(cypherTopics = this.cypherTopics + other.cypherTopics,
-                cdcSourceIdTopics = this.cdcSourceIdTopics + other.cdcSourceIdTopics,
-                cdcSchemaTopics = this.cdcSchemaTopics + other.cdcSchemaTopics,
-                cudTopics = this.cudTopics + other.cudTopics,
-                nodePatternTopics = this.nodePatternTopics + other.nodePatternTopics,
-                relPatternTopics = this.relPatternTopics + other.relPatternTopics,
-                invalid = this.invalid + other.invalid)
-    }
-
     fun allTopics(): List<String> = this.asMap()
             .map {
-                if (it.key.group == TopicTypeGroup.CDC || it.key.group == TopicTypeGroup.CUD) {
-                    (it.value as Set<String>).toList()
-                } else {
-                    (it.value as Map<String, Any>).keys.toList()
+                when (it.key.group) {
+                    TopicTypeGroup.CDC, TopicTypeGroup.CUD -> if (it.key != TopicType.CDC_SOURCE_ID) {
+                        (it.value as Set<String>).toList()
+                    } else {
+                        (it.value as Pair<Set<String>, SourceIdIngestionStrategyConfig>).first
+                    }
+                    else -> (it.value as Map<String, Any>).keys.toList()
                 }
             }
             .flatten()
@@ -43,7 +37,10 @@ data class Topics(val cypherTopics: Map<String, String> = emptyMap(),
             TopicType.PATTERN_NODE to nodePatternTopics, TopicType.PATTERN_RELATIONSHIP to relPatternTopics)
 
     companion object {
-        fun from(map: Map<String, Any?>, replacePrefix: Pair<String, String> = ("" to ""), dbName: String = "", invalidTopics: List<String> = emptyList()): Topics {
+        fun from(map: Map<String, Any?>,
+                 replacePrefix: Pair<String, String> = ("" to ""),
+                 dbName: String = "",
+                 invalidTopics: List<String> = emptyList()): Topics {
             val config = map
                     .filterKeys { if (dbName.isNotBlank()) it.toLowerCase().endsWith(".to.$dbName") else !it.contains(".to.") }
                     .mapKeys { if (dbName.isNotBlank()) it.key.replace(".to.$dbName", "", true) else it.key }
@@ -54,16 +51,25 @@ data class Topics(val cypherTopics: Map<String, String> = emptyMap(),
             val nodePatterKey = TopicType.PATTERN_NODE.replaceKeyBy(replacePrefix)
             val relPatterKey = TopicType.PATTERN_RELATIONSHIP.replaceKeyBy(replacePrefix)
             val cypherTopics = TopicUtils.filterByPrefix(config, cypherTopicPrefix)
+            val mergeNodeProperties = map[Neo4jSinkConnectorConfig.TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED]
+                .toString()
+                .toBoolean()
+            val mergeRelProperties = map[Neo4jSinkConnectorConfig.TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED]
+                .toString()
+                .toBoolean()
             val nodePatternTopics = TopicUtils
                     .filterByPrefix(config, nodePatterKey, invalidTopics)
-                    .mapValues { NodePatternConfiguration.parse(it.value) }
+                    .mapValues { NodePatternConfiguration.parse(it.value,mergeNodeProperties) }
             val relPatternTopics = TopicUtils
                     .filterByPrefix(config, relPatterKey, invalidTopics)
-                    .mapValues { RelationshipPatternConfiguration.parse(it.value) }
+                    .mapValues { RelationshipPatternConfiguration.parse(it.value, mergeNodeProperties, mergeRelProperties) }
             val cdcSourceIdTopics = TopicUtils.splitTopics(config[sourceIdKey] as? String, invalidTopics)
             val cdcSchemaTopics = TopicUtils.splitTopics(config[schemaKey] as? String, invalidTopics)
             val cudTopics = TopicUtils.splitTopics(config[cudKey] as? String, invalidTopics)
-            return Topics(cypherTopics, cdcSourceIdTopics, cdcSchemaTopics, cudTopics, nodePatternTopics, relPatternTopics)
+            val sourceIdStrategyConfig = SourceIdIngestionStrategyConfig(
+                map.getOrDefault(Neo4jSinkConnectorConfig.TOPIC_CDC_SOURCE_ID_LABEL_NAME, SourceIdIngestionStrategyConfig.DEFAULT.labelName).toString(),
+                map.getOrDefault(Neo4jSinkConnectorConfig.TOPIC_CDC_SOURCE_ID_ID_NAME, SourceIdIngestionStrategyConfig.DEFAULT.idName).toString())
+            return Topics(cypherTopics, (cdcSourceIdTopics to sourceIdStrategyConfig), cdcSchemaTopics, cudTopics, nodePatternTopics, relPatternTopics)
         }
     }
 }
@@ -104,12 +110,15 @@ object TopicUtils {
         }
     }
 
-    fun toStrategyMap(topics: Topics, sourceIdStrategyConfig: SourceIdIngestionStrategyConfig): Map<TopicType, Any> {
+    fun toStrategyMap(topics: Topics): Map<TopicType, Any> {
         return topics.asMap()
                 .filterKeys { it != TopicType.CYPHER }
                 .mapValues { (type, config) ->
                     when (type) {
-                        TopicType.CDC_SOURCE_ID -> SourceIdIngestionStrategy(sourceIdStrategyConfig)
+                        TopicType.CDC_SOURCE_ID -> {
+                            val (topics, sourceIdStrategyConfig) = (config as Pair<Set<String>, SourceIdIngestionStrategyConfig>)
+                            SourceIdIngestionStrategy(sourceIdStrategyConfig)
+                        }
                         TopicType.CDC_SCHEMA -> SchemaIngestionStrategy()
                         TopicType.CUD -> CUDIngestionStrategy()
                         TopicType.PATTERN_NODE -> {

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

@@ -45,6 +45,11 @@ neo4j.topic.cdc.schema=Type: String;\nDescription: The topic list (separated by
 neo4j.batch.parallelize=Type: Boolean;\nDescription: If enabled messages are processed concurrently in the sink. \
   Non concurrent execution supports in-order processing, e.g. for CDC (default true)
 neo4j.topic.cud=Type: String;\nDescription: The topic list (separated by semicolon) that manages CUD events
+neo4j.topic.pattern.merge.node.properties.enabled=Type: Boolean;\nDescription: If enabled nodes properties will be merged when \
+  using Sink `Node pattern` strategy (default false). In case of using Sink `Relationship pattern` strategy edge nodes properties will be merged when \
+  creating relationships (default false)
+neo4j.topic.pattern.merge.relationship.properties.enabled=Type: Boolean;\nDescription: If enabled relationships properties will be merged when creating relationships \
+  using Sink `Relationship pattern` strategy (default false)
 
 ## Source Properties
 topic=Type: String;\nDescription: The topic where the Source will publish the data

common/src/main/kotlin/streams/service/Topics.kt kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/utils/Topics.kt

@@ -57,7 +57,7 @@
         <jackson.version>2.13.4</jackson.version>
         <kotlin.compiler.incremental>true</kotlin.compiler.incremental>
         <neo4j.java.driver.version>4.4.9</neo4j.java.driver.version>
-        <testcontainers.version>1.15.1</testcontainers.version>
+        <testcontainers.version>1.17.6</testcontainers.version>
         <avro.version>1.8.2</avro.version>
         <mokito.version>4.8.0</mokito.version>
         <junit.version>4.13.2</junit.version>

kafka-connect-neo4j/src/main/resources/kafka-connect-neo4j.properties

@@ -57,7 +57,7 @@ private class DatabasesWaitStrategy(private val auth: AuthToken): AbstractWaitSt
 }
 
 class Neo4jContainerExtension(dockerImage: String): Neo4jContainer<Neo4jContainerExtension>(dockerImage) {
-    constructor(): this("neo4j:4.4.10-enterprise")
+    constructor(): this("neo4j:5.1-enterprise")
     private val logger = LoggerFactory.getLogger(Neo4jContainerExtension::class.java)
     var driver: Driver? = null
     var session: Session? = null
@@ -100,7 +100,7 @@ class Neo4jContainerExtension(dockerImage: String): Neo4jContainer<Neo4jContaine
     }
 
     fun withKafka(kafka: KafkaContainer): Neo4jContainerExtension {
-        return withKafka(kafka.network, kafka.networkAliases.map { "$it:9092" }.joinToString(","))
+        return withKafka(kafka.network!!, kafka.networkAliases.map { "$it:9092" }.joinToString(","))
     }
 
     fun withKafka(network: Network, bootstrapServers: String): Neo4jContainerExtension {