Fixes #293: streams.sink.polling.interval is possibly confusing and may not be used anyway (#299)

Author: mroiter-larus

consumer/src/main/kotlin/streams/StreamsSinkConfiguration.kt

@@ -8,7 +8,6 @@ import streams.service.sink.strategy.SourceIdIngestionStrategyConfig
 
 data class StreamsSinkConfiguration(val enabled: Boolean = StreamsConfig.SINK_ENABLED_VALUE,
                                     val proceduresEnabled: Boolean = StreamsConfig.PROCEDURES_ENABLED_VALUE,
-                                    val sinkPollingInterval: Long = 10000,
                                     val topics: Topics = Topics(),
                                     val errorConfig: Map<String,Any?> = emptyMap(),
                                     val sourceIdStrategyConfig: SourceIdIngestionStrategyConfig = SourceIdIngestionStrategyConfig()) {
@@ -42,7 +41,6 @@ data class StreamsSinkConfiguration(val enabled: Boolean = StreamsConfig.SINK_EN
 
             return default.copy(enabled = cfg.isSinkEnabled(dbName),
                     proceduresEnabled = cfg.hasProceduresEnabled(dbName),
-                    sinkPollingInterval = cfg.config.getOrDefault("streams.sink.polling.interval", default.sinkPollingInterval).toString().toLong(),
                     topics = topics,
                     errorConfig = errorHandler,
                     sourceIdStrategyConfig = sourceIdStrategyConfig)

consumer/src/test/kotlin/streams/StreamsSinkConfigurationTest.kt

@@ -38,15 +38,13 @@ class StreamsSinkConfigurationTest {
 
     @Test
     fun `should manage only topics for default db`() {
-        val pollingInterval = "10"
         val topicKey = "streams.sink.topic.cypher.myTopic"
         val topicValue = "MERGE (n:Label{ id: event.id })"
         val topicKeyNeo = "streams.sink.topic.cypher.myTopicNeo.to.neo4j"
         val topicValueNeo = "MERGE (n:Neo4j{ id: event.id })"
         val topicKeyFoo = "streams.sink.topic.cypher.myTopicFoo.to.foo"
         val topicValueFoo = "MERGE (n:Foo{ id: event.id })"
-        val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                topicKey to topicValue,
+        val config = mapOf(topicKey to topicValue,
                 topicKeyNeo to topicValueNeo,
                 topicKeyFoo to topicValueFoo)
         streamsConfig.config.putAll(config)
@@ -57,15 +55,13 @@ class StreamsSinkConfigurationTest {
 
     @Test
     fun `should manage only topics for non default db`() {
-        val pollingInterval = "10"
         val topicKey = "streams.sink.topic.cypher.myTopic"
         val topicValue = "MERGE (n:Label{ id: event.id })"
         val topicKeyNeo = "streams.sink.topic.cypher.myTopicNeo.to.neo4j"
         val topicValueNeo = "MERGE (n:Neo4j{ id: event.id })"
         val topicKeyFoo = "streams.sink.topic.cypher.myTopicFoo.to.foo"
         val topicValueFoo = "MERGE (n:Foo{ id: event.id })"
-        val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                topicKey to topicValue,
+        val config = mapOf(topicKey to topicValue,
                 topicKeyNeo to topicValueNeo,
                 topicKeyFoo to topicValueFoo)
         streamsConfig.config.putAll(config)
@@ -82,23 +78,20 @@ class StreamsSinkConfigurationTest {
 
     @Test
     fun shouldReturnConfigurationFromMap() {
-        val pollingInterval = "10"
         val topic = "topic-neo"
         val cdctopic = "cdctopic"
         val topicKey = "streams.sink.topic.cypher.$topic"
         val topicValue = "MERGE (n:Label{ id: event.id }) "
         val customLabel = "CustomLabel"
         val customId = "customId"
-        val config = mapOf(
-                "streams.sink.polling.interval" to pollingInterval,
-                topicKey to topicValue,
+        val config = mapOf(topicKey to topicValue,
                 "streams.sink.enabled" to "false",
                 "streams.sink.topic.cdc.sourceId" to cdctopic,
                 "streams.sink.topic.cdc.sourceId.labelName" to customLabel,
                 "streams.sink.topic.cdc.sourceId.idName" to customId)
         streamsConfig.config.putAll(config)
         val streamsSinkConf = StreamsSinkConfiguration.from(streamsConfig, defalutDbName)
-        testFromConf(streamsSinkConf, pollingInterval, topic, topicValue)
+        testFromConf(streamsSinkConf, topic, topicValue)
         assertFalse { streamsSinkConf.enabled }
         assertEquals(setOf(cdctopic), streamsSinkConf.topics.asMap()[TopicType.CDC_SOURCE_ID])
         assertEquals(customLabel, streamsSinkConf.sourceIdStrategyConfig.labelName)
@@ -107,12 +100,10 @@ class StreamsSinkConfigurationTest {
 
     @Test(expected = TopicValidationException::class)
     fun shouldFailWithCrossDefinedTopics() {
-        val pollingInterval = "10"
         val topic = "topic-neo"
         val topicKey = "streams.sink.topic.cypher.$topic"
         val topicValue = "MERGE (n:Label{ id: event.id }) "
-        val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                topicKey to topicValue,
+        val config = mapOf(topicKey to topicValue,
                 "streams.sink.topic.pattern.node.nodePatternTopic" to "User{!userId,name,surname,address.city}",
                 "streams.sink.enabled" to "false",
                 "streams.sink.topic.cdc.sourceId" to topic)
@@ -122,10 +113,8 @@ class StreamsSinkConfigurationTest {
 
     @Test(expected = TopicValidationException::class)
     fun shouldFailWithCrossDefinedCDCTopics() {
-        val pollingInterval = "10"
         val topic = "topic-neo"
-        val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                "streams.sink.enabled" to "false",
+        val config = mapOf("streams.sink.enabled" to "false",
                 "streams.sink.topic.cdc.sourceId" to topic,
                 "streams.sink.topic.cdc.schema" to topic)
         streamsConfig.config.putAll(config)
@@ -135,10 +124,8 @@ class StreamsSinkConfigurationTest {
     companion object {
         fun testDefaultConf(default: StreamsSinkConfiguration) {
             assertEquals(emptyMap(), default.topics.cypherTopics)
-            assertEquals(10000, default.sinkPollingInterval)
         }
-        fun testFromConf(streamsConfig: StreamsSinkConfiguration, pollingInterval: String, topic: String, topicValue: String) {
-            assertEquals(pollingInterval.toLong(), streamsConfig.sinkPollingInterval)
+        fun testFromConf(streamsConfig: StreamsSinkConfiguration, topic: String, topicValue: String) {
             assertEquals(1, streamsConfig.topics.cypherTopics.size)
             assertEquals(topicValue, streamsConfig.topics.cypherTopics[topic])
         }

consumer/src/test/kotlin/streams/kafka/KafkaSinkConfigurationTest.kt

@@ -57,7 +57,6 @@ class KafkaSinkConfigurationTest {
 
     @Test
     fun `should return configuration from map`() {
-        val pollingInterval = "10"
         val topic = "topic-neo"
         val topicKey = "streams.sink.topic.cypher.$topic"
         val topicValue = "MERGE (n:Label{ id: event.id }) "
@@ -66,8 +65,7 @@ class KafkaSinkConfigurationTest {
         val group = "foo"
         val autoOffsetReset = "latest"
         val autoCommit = "false"
-        val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                topicKey to topicValue,
+        val config = mapOf(topicKey to topicValue,
                 "kafka.zookeeper.connect" to zookeeper,
                 "kafka.bootstrap.servers" to bootstrap,
                 "kafka.auto.offset.reset" to autoOffsetReset,
@@ -84,7 +82,7 @@ class KafkaSinkConfigurationTest {
 
         streamsConfig.config.putAll(config)
         val kafkaSinkConfiguration = KafkaSinkConfiguration.create(streamsConfig, defaultDbName)
-        StreamsSinkConfigurationTest.testFromConf(kafkaSinkConfiguration.streamsSinkConfiguration, pollingInterval, topic, topicValue)
+        StreamsSinkConfigurationTest.testFromConf(kafkaSinkConfiguration.streamsSinkConfiguration, topic, topicValue)
         assertEquals(emptyMap(), kafkaSinkConfiguration.extraProperties)
         assertEquals(zookeeper, kafkaSinkConfiguration.zookeeperConnect)
         assertEquals(bootstrap, kafkaSinkConfiguration.bootstrapServers)
@@ -97,15 +95,13 @@ class KafkaSinkConfigurationTest {
         assertEquals(expectedMap, resultMap)
 
         val streamsConfig = StreamsSinkConfiguration.from(streamsConfig, defaultDbName)
-        assertEquals(pollingInterval.toLong(), streamsConfig.sinkPollingInterval)
         assertTrue { streamsConfig.topics.cypherTopics.containsKey(topic) }
         assertEquals(topicValue, streamsConfig.topics.cypherTopics[topic])
     }
 
     @Test
     fun `should return configuration from map for non default DB`() {
         val dbName = "foo"
-        val pollingInterval = "10"
         val topic = "topic-neo"
         val topicKey = "streams.sink.topic.cypher.$topic"
         val topicValue = "MERGE (n:Label{ id: event.id })"
@@ -116,8 +112,7 @@ class KafkaSinkConfigurationTest {
         val group = "mygroup"
         val autoOffsetReset = "latest"
         val autoCommit = "false"
-        val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                topicKey to topicValue,
+        val config = mapOf(topicKey to topicValue,
                 topicKeyFoo to topicValueFoo,
                 "kafka.zookeeper.connect" to zookeeper,
                 "kafka.bootstrap.servers" to bootstrap,
@@ -135,7 +130,7 @@ class KafkaSinkConfigurationTest {
 
         streamsConfig.config.putAll(config)
         val kafkaSinkConfiguration = KafkaSinkConfiguration.create(streamsConfig, dbName)
-        StreamsSinkConfigurationTest.testFromConf(kafkaSinkConfiguration.streamsSinkConfiguration, pollingInterval, topic, topicValueFoo)
+        StreamsSinkConfigurationTest.testFromConf(kafkaSinkConfiguration.streamsSinkConfiguration, topic, topicValueFoo)
         assertEquals(emptyMap(), kafkaSinkConfiguration.extraProperties)
         assertEquals(zookeeper, kafkaSinkConfiguration.zookeeperConnect)
         assertEquals(bootstrap, kafkaSinkConfiguration.bootstrapServers)
@@ -148,7 +143,6 @@ class KafkaSinkConfigurationTest {
         assertEquals(expectedMap, resultMap)
 
         val streamsConfig = StreamsSinkConfiguration.from(streamsConfig, dbName)
-        assertEquals(pollingInterval.toLong(), streamsConfig.sinkPollingInterval)
         assertEquals(1, streamsConfig.topics.cypherTopics.size)
         assertTrue { streamsConfig.topics.cypherTopics.containsKey(topic) }
         assertEquals(topicValueFoo, streamsConfig.topics.cypherTopics[topic])
@@ -160,15 +154,13 @@ class KafkaSinkConfigurationTest {
         val zookeeper = "zookeeper:2181"
         val bootstrap = "bootstrap:9092"
         try {
-            val pollingInterval = "10"
             val topic = "topic-neo"
             val topicKey = "streams.sink.topic.cypher.$topic"
             val topicValue = "MERGE (n:Label{ id: event.id }) "
             val group = "foo"
             val autoOffsetReset = "latest"
             val autoCommit = "false"
-            val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                    topicKey to topicValue,
+            val config = mapOf(topicKey to topicValue,
                     "$topicKey.to.foo" to "$topicValue SET n += event.properties",
                     "kafka.zookeeper.connect" to zookeeper,
                     "kafka.bootstrap.servers" to bootstrap,
@@ -190,15 +182,13 @@ class KafkaSinkConfigurationTest {
         val zookeeper = "zookeeper:2181"
         val bootstrap = ""
         try {
-            val pollingInterval = "10"
             val topic = "topic-neo"
             val topicKey = "streams.sink.topic.cypher.$topic"
             val topicValue = "MERGE (n:Label{ id: event.id }) "
             val group = "foo"
             val autoOffsetReset = "latest"
             val autoCommit = "false"
-            val config = mapOf("streams.sink.polling.interval" to pollingInterval,
-                    topicKey to topicValue,
+            val config = mapOf(topicKey to topicValue,
                     "kafka.zookeeper.connect" to zookeeper,
                     "kafka.bootstrap.servers" to bootstrap,
                     "kafka.auto.offset.reset" to autoOffsetReset,

doc/asciidoc/docker/index.adoc

@@ -222,7 +222,7 @@ NEO4J_streams_source_schema_polling_interval: 10000
 ----
 
 this means that every 10 seconds the Streams plugin polls the DB in order to retrieve schema changes and store them.
-So after you created the indexes you need almost to wait 10 seconds before the next step, otherwise the
+So after you created the indexes you need almost to wait 10 seconds before the next step.
 
 Now lets go to the `Source` and, in order to import the Stackoverflow dataset, execute the following query:
 

doc/asciidoc/faq/index.adoc

@@ -30,7 +30,9 @@ If you decide to use the Neo4j Streams plugin, then the Neo4j source instance wi
 [source, ini]
 ----
 streams.sink.enabled=false
-streams.source.schema.polling.interval=10000
+streams.source.enabled=true
+streams.source.topic.nodes.<TOPIC_NAME>=<PATTERN>
+streams.source.topic.relationships.<TOPIC_NAME>=<PATTERN>
 ----
 
 and the Neo4j sink instance will be configured as follow:

doc/asciidoc/kafka-ssl/index.adoc

@@ -83,7 +83,6 @@ Note that the passwords are stored in plaintext so limit access to this `neo4j.c
 kafka.zookeeper.connect=xxx.xxx.xxx.xxx:2181
 kafka.bootstrap.servers=xxx.xxx.xxx.xxx:9094
 streams.sink.enabled=false
-streams.sink.polling.interval=1000
 
 streams.source.topic.nodes.neoTest=Person{*}