refactor: make sink config consistent with the source (#48)

Author: venikkin

common/src/main/kotlin/org/neo4j/connectors/kafka/service/StreamsSinkService.kt

@@ -18,24 +18,20 @@ package org.neo4j.connectors.kafka.service
 
 import org.neo4j.connectors.kafka.service.sink.strategy.IngestionStrategy
 
-const val STREAMS_TOPIC_KEY: String = "streams.sink.topic"
-const val STREAMS_TOPIC_CDC_KEY: String = "streams.sink.topic.cdc"
-
 enum class TopicTypeGroup {
   CYPHER,
   CDC,
   PATTERN,
   CUD
 }
 
-enum class TopicType(val group: TopicTypeGroup, val key: String) {
-  CDC_SOURCE_ID(group = TopicTypeGroup.CDC, key = "$STREAMS_TOPIC_CDC_KEY.sourceId"),
-  CYPHER(group = TopicTypeGroup.CYPHER, key = "$STREAMS_TOPIC_KEY.cypher"),
-  PATTERN_NODE(group = TopicTypeGroup.PATTERN, key = "$STREAMS_TOPIC_KEY.pattern.node"),
-  PATTERN_RELATIONSHIP(
-      group = TopicTypeGroup.PATTERN, key = "$STREAMS_TOPIC_KEY.pattern.relationship"),
-  CDC_SCHEMA(group = TopicTypeGroup.CDC, key = "$STREAMS_TOPIC_CDC_KEY.schema"),
-  CUD(group = TopicTypeGroup.CUD, key = "$STREAMS_TOPIC_KEY.cud")
+enum class TopicType(val group: TopicTypeGroup) {
+  CDC_SOURCE_ID(group = TopicTypeGroup.CDC),
+  CYPHER(group = TopicTypeGroup.CYPHER),
+  PATTERN_NODE(group = TopicTypeGroup.PATTERN),
+  PATTERN_RELATIONSHIP(group = TopicTypeGroup.PATTERN),
+  CDC_SCHEMA(group = TopicTypeGroup.CDC),
+  CUD(group = TopicTypeGroup.CUD)
 }
 
 data class StreamsSinkEntity(val key: Any?, val value: Any?)

common/src/main/kotlin/org/neo4j/connectors/kafka/utils/PropertiesUtil.kt

@@ -24,36 +24,35 @@ class PropertiesUtil {
   companion object {
     private val LOGGER = LoggerFactory.getLogger(PropertiesUtil::class.java)
     private const val DEFAULT_VERSION = "unknown"
-    private var properties: Properties? = null
-    private var VERSION: String? = null
+    private var properties: Properties = Properties()
+    private var VERSION: String
 
     init {
-      properties = Properties()
-      properties!!.load(
+      properties.load(
           PropertiesUtil::class.java.getResourceAsStream("/neo4j-configuration.properties"))
-      properties!!.load(
+      properties.load(
           PropertiesUtil::class.java.getResourceAsStream("/neo4j-source-configuration.properties"))
-      properties!!.load(
+      properties.load(
           PropertiesUtil::class.java.getResourceAsStream("/neo4j-sink-configuration.properties"))
-      properties!!.load(
+      properties.load(
           PropertiesUtil::class.java.getResourceAsStream("/kafka-connect-version.properties"))
-      properties!!.load(
+      properties.load(
           PropertiesUtil::class.java.getResourceAsStream("/kafka-connect-neo4j.properties"))
       VERSION =
           try {
-            properties!!.getProperty("version", DEFAULT_VERSION).trim()
+            properties.getProperty("version", DEFAULT_VERSION).trim()
           } catch (e: Exception) {
             LOGGER.warn("error while loading version:", e)
             DEFAULT_VERSION
           }
     }
 
     fun getVersion(): String {
-      return VERSION!!
+      return VERSION
     }
 
     fun getProperty(key: String): String {
-      return properties!!.getProperty(key)
+      return properties.getProperty(key)
     }
   }
 }

common/src/main/resources/neo4j-sink-configuration.properties

@@ -13,17 +13,17 @@
 # limitations under the License.
 ##
 ## Connection Properties
-neo4j.topic.cdc.sourceId=Type: String;\nDescription: The topic list (separated by semicolon) that manages CDC events with the `SourceId` strategy
-neo4j.topic.cdc.sourceId.labelName=Type: String;\nDescription: The label name attached to the events with the `SourceId` strategy (default SourceEvent)
-neo4j.topic.cdc.sourceId.idName=Type: String;\nDescription: The id property name attached to the events with the `SourceId` strategy (default sourceId)
-neo4j.topic.cdc.schema=Type: String;\nDescription: The topic list (separated by semicolon) that manages CDC events with the `Schema` strategy
+neo4j.cdc.sourceId.topics=Type: String;\nDescription: The topic list (separated by semicolon) that manages CDC events with the `SourceId` strategy
+neo4j.cdc.sourceId.labelName=Type: String;\nDescription: The label name attached to the events with the `SourceId` strategy (default SourceEvent)
+neo4j.cdc.sourceId.idName=Type: String;\nDescription: The id property name attached to the events with the `SourceId` strategy (default sourceId)
+neo4j.cdc.schema.topics=Type: String;\nDescription: The topic list (separated by semicolon) that manages CDC events with the `Schema` strategy
 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=Type: Boolean;\nDescription: If enabled nodes properties will be merged when \
+neo4j.cud.topics=Type: String;\nDescription: The topic list (separated by semicolon) that manages CUD events
+neo4j.pattern.node.merge-properties=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=Type: Boolean;\nDescription: If enabled relationships properties will be merged when creating relationships \
+neo4j.pattern.relationship.merge-properties=Type: Boolean;\nDescription: If enabled relationships properties will be merged when creating relationships \
   using Sink `Relationship pattern` strategy (default false)
 neo4j.batch-size=Type: Integer;\nDescription: Max number of messages processed per batch.
 neo4j.batch-timeout=Type: Duration;\nDescription: Maximum amount of time a batch is allowed to be processed.

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/DeprecatedNeo4jSinkConfiguration.kt

@@ -33,20 +33,28 @@ class DeprecatedNeo4jSinkConfiguration(originals: Map<*, *>) :
     @Deprecated("deprecated in favour of ${SinkConfiguration.BATCH_PARALLELIZE}")
     const val BATCH_PARALLELIZE = "neo4j.batch.parallelize"
 
+    @Deprecated("deprecated in favour of ${SinkConfiguration.CYPHER_TOPIC_PREFIX}")
     const val TOPIC_CYPHER_PREFIX = "neo4j.topic.cypher."
+    @Deprecated("deprecated in favour of ${SinkConfiguration.CDC_SOURCE_ID_TOPICS}")
     const val TOPIC_CDC_SOURCE_ID = "neo4j.topic.cdc.sourceId"
+    @Deprecated("deprecated in favour of ${SinkConfiguration.CDC_SOURCE_ID_LABEL_NAME}")
     const val TOPIC_CDC_SOURCE_ID_LABEL_NAME = "neo4j.topic.cdc.sourceId.labelName"
+    @Deprecated("deprecated in favour of ${SinkConfiguration.CDC_SOURCE_ID_ID_NAME}")
     const val TOPIC_CDC_SOURCE_ID_ID_NAME = "neo4j.topic.cdc.sourceId.idName"
+    @Deprecated("deprecated in favour of ${SinkConfiguration.PATTERN_NODE_TOPIC_PREFIX}")
     const val TOPIC_PATTERN_NODE_PREFIX = "neo4j.topic.pattern.node."
+    @Deprecated("deprecated in favour of ${SinkConfiguration.PATTERN_RELATIONSHIP_TOPIC_PREFIX}")
     const val TOPIC_PATTERN_RELATIONSHIP_PREFIX = "neo4j.topic.pattern.relationship."
-    @Deprecated("deprecated in favour of ${SinkConfiguration.TOPIC_PATTERN_MERGE_NODE_PROPERTIES}")
+    @Deprecated("deprecated in favour of ${SinkConfiguration.PATTERN_NODE_MERGE_PROPERTIES}")
     const val TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED =
         "neo4j.topic.pattern.merge.node.properties.enabled"
     @Deprecated(
-        "deprecated in favour of ${SinkConfiguration.TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES}")
+        "deprecated in favour of ${SinkConfiguration.PATTERN_RELATIONSHIP_MERGE_PROPERTIES}")
     const val TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED =
         "neo4j.topic.pattern.merge.relationship.properties.enabled"
+    @Deprecated("deprecated in favour of ${SinkConfiguration.CDC_SCHEMA_TOPICS}")
     const val TOPIC_CDC_SCHEMA = "neo4j.topic.cdc.schema"
+    @Deprecated("deprecated in favour of ${SinkConfiguration.CUD_TOPICS}")
     const val TOPIC_CUD = "neo4j.topic.cud"
 
     const val DEFAULT_BATCH_PARALLELIZE = true

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/SinkConfiguration.kt

@@ -51,7 +51,7 @@ class SinkConfiguration(originals: Map<*, *>) :
   val batchTimeout
     get(): Duration = Duration.parseSimpleString(getString(BATCH_TIMEOUT))
 
-  val topics: Topics by lazy { Topics.from(originals(), "streams.sink." to "neo4j.") }
+  val topics: Topics by lazy { Topics.from(originals()) }
 
   val strategyMap: Map<TopicType, Any> by lazy { TopicUtils.toStrategyMap(topics) }
 
@@ -83,42 +83,62 @@ class SinkConfiguration(originals: Map<*, *>) :
     const val BATCH_TIMEOUT = "neo4j.batch-timeout"
     const val BATCH_PARALLELIZE = "neo4j.batch-parallelize"
 
-    const val TOPIC_CYPHER_PREFIX = "neo4j.topic.cypher."
-    const val TOPIC_CDC_SOURCE_ID = "neo4j.topic.cdc.sourceId"
-    const val TOPIC_CDC_SOURCE_ID_LABEL_NAME = "neo4j.topic.cdc.sourceId.labelName"
-    const val TOPIC_CDC_SOURCE_ID_ID_NAME = "neo4j.topic.cdc.sourceId.idName"
-    const val TOPIC_CDC_SCHEMA = "neo4j.topic.cdc.schema"
-    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 = "neo4j.topic.pattern.merge-node-properties"
-    const val TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES =
-        "neo4j.topic.pattern.merge-relationship-properties"
-    const val TOPIC_CUD = "neo4j.topic.cud"
+    const val CYPHER_TOPIC_PREFIX = "neo4j.cypher.topic."
+    const val CDC_SOURCE_ID_TOPICS = "neo4j.cdc.sourceId.topics"
+    const val CDC_SOURCE_ID_LABEL_NAME = "neo4j.cdc.sourceId.labelName"
+    const val CDC_SOURCE_ID_ID_NAME = "neo4j.cdc.sourceId.idName"
+    const val CDC_SCHEMA_TOPICS = "neo4j.cdc.schema.topics"
+    const val PATTERN_NODE_TOPIC_PREFIX = "neo4j.pattern.node.topic."
+    const val PATTERN_RELATIONSHIP_TOPIC_PREFIX = "neo4j.pattern.relationship.topic."
+    const val PATTERN_NODE_MERGE_PROPERTIES = "neo4j.pattern.node.merge-properties"
+    const val PATTERN_RELATIONSHIP_MERGE_PROPERTIES = "neo4j.pattern.relationship.merge-properties"
+    const val CUD_TOPICS = "neo4j.cud.topics"
 
     const val DEFAULT_BATCH_SIZE = 1000
     val DEFAULT_BATCH_TIMEOUT = 0.seconds
     const val DEFAULT_BATCH_PARALLELIZE = true
     const val DEFAULT_TOPIC_PATTERN_MERGE_NODE_PROPERTIES = false
     const val DEFAULT_TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES = false
 
+    @JvmStatic
+    val KEY_REPLACEMENTS =
+        mapOf(
+            DeprecatedNeo4jSinkConfiguration.TOPIC_CYPHER_PREFIX to CYPHER_TOPIC_PREFIX,
+            DeprecatedNeo4jSinkConfiguration.TOPIC_PATTERN_NODE_PREFIX to PATTERN_NODE_TOPIC_PREFIX,
+            DeprecatedNeo4jSinkConfiguration.TOPIC_PATTERN_RELATIONSHIP_PREFIX to
+                PATTERN_RELATIONSHIP_TOPIC_PREFIX)
+
     fun migrateSettings(oldSettings: Map<String, Any>): Map<String, String> {
-      val migrated = Neo4jConfiguration.migrateSettings(oldSettings, true).toMutableMap()
+      val migratedBase = Neo4jConfiguration.migrateSettings(oldSettings, false)
+      val migrated = HashMap<String, String>(migratedBase.size)
 
-      oldSettings.forEach {
+      migratedBase.forEach {
         when (it.key) {
-          DeprecatedNeo4jConfiguration.BATCH_SIZE -> migrated[BATCH_SIZE] = it.value.toString()
+          DeprecatedNeo4jConfiguration.BATCH_SIZE -> migrated[BATCH_SIZE] = it.value
           DeprecatedNeo4jConfiguration.BATCH_TIMEOUT_MSECS ->
               migrated[BATCH_TIMEOUT] = "${it.value}ms"
           DeprecatedNeo4jSinkConfiguration.BATCH_PARALLELIZE ->
-              migrated[BATCH_PARALLELIZE] = it.value.toString()
+              migrated[BATCH_PARALLELIZE] = it.value
           DeprecatedNeo4jSinkConfiguration.TOPIC_PATTERN_MERGE_NODE_PROPERTIES_ENABLED ->
-              migrated[TOPIC_PATTERN_MERGE_NODE_PROPERTIES] = it.value.toString()
+              migrated[PATTERN_NODE_MERGE_PROPERTIES] = it.value
           DeprecatedNeo4jSinkConfiguration.TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES_ENABLED ->
-              migrated[TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES] = it.value.toString()
-          else ->
-              if (!migrated.containsKey(it.key)) {
-                migrated[it.key] = it.value.toString()
-              }
+              migrated[PATTERN_RELATIONSHIP_MERGE_PROPERTIES] = it.value
+          DeprecatedNeo4jSinkConfiguration.TOPIC_CDC_SOURCE_ID ->
+              migrated[CDC_SOURCE_ID_TOPICS] = it.value.replaceLegacyDelimiter()
+          DeprecatedNeo4jSinkConfiguration.TOPIC_CDC_SOURCE_ID_LABEL_NAME ->
+              migrated[CDC_SOURCE_ID_LABEL_NAME] = it.value
+          DeprecatedNeo4jSinkConfiguration.TOPIC_CDC_SOURCE_ID_ID_NAME ->
+              migrated[CDC_SOURCE_ID_ID_NAME] = it.value
+          DeprecatedNeo4jSinkConfiguration.TOPIC_CDC_SCHEMA ->
+              migrated[CDC_SCHEMA_TOPICS] = it.value.replaceLegacyDelimiter()
+          DeprecatedNeo4jSinkConfiguration.TOPIC_CUD ->
+              migrated[CUD_TOPICS] = it.value.replaceLegacyDelimiter()
+          else -> {
+            val migratedKey = replaceLegacyPropertyKeys(it.key)
+            if (!migrated.containsKey(migratedKey)) {
+              migrated[migratedKey] = it.value
+            }
+          }
         }
       }
 
@@ -132,64 +152,62 @@ class SinkConfiguration(originals: Map<*, *>) :
     fun config(): ConfigDef =
         Neo4jConfiguration.config()
             .define(
-                ConfigKeyBuilder.of(TOPIC_CDC_SOURCE_ID, ConfigDef.Type.LIST) {
+                ConfigKeyBuilder.of(CDC_SOURCE_ID_TOPICS, ConfigDef.Type.LIST) {
                   importance = ConfigDef.Importance.HIGH
                   defaultValue = ""
                   group = ConfigGroup.TOPIC_CYPHER_MAPPING
                 })
             .define(
-                ConfigKeyBuilder.of(TOPIC_CDC_SOURCE_ID_LABEL_NAME, ConfigDef.Type.STRING) {
+                ConfigKeyBuilder.of(CDC_SOURCE_ID_LABEL_NAME, ConfigDef.Type.STRING) {
                   importance = ConfigDef.Importance.HIGH
                   defaultValue = SourceIdIngestionStrategyConfig.DEFAULT.labelName
                   group = ConfigGroup.TOPIC_CYPHER_MAPPING
                   recommender =
-                      Recommenders.visibleIfNotEmpty(Predicate.isEqual(TOPIC_CDC_SOURCE_ID))
+                      Recommenders.visibleIfNotEmpty(Predicate.isEqual(CDC_SOURCE_ID_TOPICS))
                 })
             .define(
-                ConfigKeyBuilder.of(TOPIC_CDC_SOURCE_ID_ID_NAME, ConfigDef.Type.STRING) {
+                ConfigKeyBuilder.of(CDC_SOURCE_ID_ID_NAME, ConfigDef.Type.STRING) {
                   importance = ConfigDef.Importance.HIGH
                   defaultValue = SourceIdIngestionStrategyConfig.DEFAULT.idName
                   group = ConfigGroup.TOPIC_CYPHER_MAPPING
                   recommender =
-                      Recommenders.visibleIfNotEmpty(Predicate.isEqual(TOPIC_CDC_SOURCE_ID))
+                      Recommenders.visibleIfNotEmpty(Predicate.isEqual(CDC_SOURCE_ID_TOPICS))
                 })
             .define(
-                ConfigKeyBuilder.of(TOPIC_CDC_SCHEMA, ConfigDef.Type.LIST) {
+                ConfigKeyBuilder.of(CDC_SCHEMA_TOPICS, ConfigDef.Type.LIST) {
                   importance = ConfigDef.Importance.HIGH
                   defaultValue = ""
                   group = ConfigGroup.TOPIC_CYPHER_MAPPING
                 })
             .define(
-                ConfigKeyBuilder.of(TOPIC_CUD, ConfigDef.Type.LIST) {
+                ConfigKeyBuilder.of(CUD_TOPICS, ConfigDef.Type.LIST) {
                   importance = ConfigDef.Importance.HIGH
                   defaultValue = ""
                   group = ConfigGroup.TOPIC_CYPHER_MAPPING
                 })
             .define(
-                ConfigKeyBuilder.of(TOPIC_PATTERN_MERGE_NODE_PROPERTIES, ConfigDef.Type.BOOLEAN) {
+                ConfigKeyBuilder.of(PATTERN_NODE_MERGE_PROPERTIES, ConfigDef.Type.BOOLEAN) {
                   importance = ConfigDef.Importance.MEDIUM
                   defaultValue = DEFAULT_TOPIC_PATTERN_MERGE_NODE_PROPERTIES
                   group = ConfigGroup.TOPIC_CYPHER_MAPPING
                   recommender =
                       Recommenders.visibleIfNotEmpty { k ->
-                        k.startsWith(TOPIC_PATTERN_NODE_PREFIX) ||
-                            k.startsWith(TOPIC_PATTERN_RELATIONSHIP_PREFIX)
+                        k.startsWith(PATTERN_NODE_TOPIC_PREFIX) ||
+                            k.startsWith(PATTERN_RELATIONSHIP_TOPIC_PREFIX)
                       }
                 })
             .define(
-                ConfigKeyBuilder.of(
-                    TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES, ConfigDef.Type.BOOLEAN) {
-                      documentation =
-                          PropertiesUtil.getProperty(TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES)
-                      importance = ConfigDef.Importance.MEDIUM
-                      defaultValue = DEFAULT_TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES
-                      group = ConfigGroup.TOPIC_CYPHER_MAPPING
-                      recommender =
-                          Recommenders.visibleIfNotEmpty { k ->
-                            k.startsWith(TOPIC_PATTERN_NODE_PREFIX) ||
-                                k.startsWith(TOPIC_PATTERN_RELATIONSHIP_PREFIX)
-                          }
-                    })
+                ConfigKeyBuilder.of(PATTERN_RELATIONSHIP_MERGE_PROPERTIES, ConfigDef.Type.BOOLEAN) {
+                  documentation = PropertiesUtil.getProperty(PATTERN_RELATIONSHIP_MERGE_PROPERTIES)
+                  importance = ConfigDef.Importance.MEDIUM
+                  defaultValue = DEFAULT_TOPIC_PATTERN_MERGE_RELATIONSHIP_PROPERTIES
+                  group = ConfigGroup.TOPIC_CYPHER_MAPPING
+                  recommender =
+                      Recommenders.visibleIfNotEmpty { k ->
+                        k.startsWith(PATTERN_NODE_TOPIC_PREFIX) ||
+                            k.startsWith(PATTERN_RELATIONSHIP_TOPIC_PREFIX)
+                      }
+                })
             .define(
                 ConfigKeyBuilder.of(BATCH_SIZE, ConfigDef.Type.INT) {
                   importance = ConfigDef.Importance.HIGH
@@ -208,5 +226,12 @@ class SinkConfiguration(originals: Map<*, *>) :
                   defaultValue = DEFAULT_BATCH_PARALLELIZE
                   group = ConfigGroup.BATCH
                 })
+
+    private fun replaceLegacyPropertyKeys(key: String) =
+        KEY_REPLACEMENTS.entries.fold(key) { k, replacement ->
+          k.replace(replacement.key, replacement.value)
+        }
+
+    private fun String.replaceLegacyDelimiter() = this.replace(';', ',')
   }
 }