fixes 160: change the streams.sink.enabled to false (#176)

Author: conker84

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

@@ -11,7 +11,7 @@ private object StreamsSinkConfigurationConstants {
     const val PROCEDURES_ENABLED = "procedures.enabled"
 }
 
-data class StreamsSinkConfiguration(val enabled: Boolean = true,
+data class StreamsSinkConfiguration(val enabled: Boolean = false,
                                     val proceduresEnabled: Boolean = true,
                                     val sinkPollingInterval: Long = 10000,
                                     val topics: Map<String, String> = emptyMap()) {

consumer/src/test/kotlin/integrations/KafkaEventSinkIT.kt

@@ -84,6 +84,7 @@ class KafkaEventSinkIT {
         var graphDatabaseBuilder = TestGraphDatabaseFactory()
                 .newImpermanentDatabaseBuilder()
                 .setConfig("kafka.bootstrap.servers", kafka.bootstrapServers)
+                .setConfig("streams.sink.enabled", "true")
         if (!testName.methodName.endsWith(EXCLUDE_LOAD_TOPIC_METHOD_SUFFIX)) {
             graphDatabaseBuilder.setConfig("streams.sink.topic.cypher.shouldWriteCypherQuery", cypherQueryTemplate)
         }

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

@@ -23,11 +23,11 @@ class StreamsSinkConfigurationTest {
         val config = Config.builder()
                 .withSetting("streams.sink.polling.interval", pollingInterval)
                 .withSetting(topicKey, topicValue)
-                .withSetting("streams.sink.enabled", "false")
+                .withSetting("streams.sink.enabled", "true")
                 .build()
         val streamsConfig = StreamsSinkConfiguration.from(config)
         testFromConf(streamsConfig, pollingInterval, topic, topicValue)
-        assertFalse { streamsConfig.enabled }
+        assertTrue { streamsConfig.enabled }
     }
 
     companion object {

doc/asciidoc/consumer/configuration.adoc

@@ -9,7 +9,6 @@ kafka.group.id=neo4j
 
 streams.sink.polling.interval=<The time, in milliseconds, spent waiting in poll if data is not available in the buffer. default=Long.MAX_VALUE>
 streams.sink.topic.cypher.<TOPIC_NAME>=<CYPHER_QUERY>
-streams.sink.enabled=<true/false, default=true>
-----
+streams.sink.enabled=<true/false, default=false>
 
 See the https://kafka.apache.org/documentation/#brokerconfigs[Apache Kafka documentation] for details on these settings.

doc/asciidoc/docker/index.adoc

@@ -68,6 +68,7 @@ In case of you are using the Sink you can define your topic/cypher-query combina
 [source,yaml]
 ----
     environment:
+      NEO4J_streams_sink_enabled: "true"
       NEO4J_streams_sink_topic_neo4j:
         "WITH event.value.payload AS payload, event.value.meta AS meta
         FOREACH (ignoreMe IN CASE WHEN payload.type = 'node' AND meta.operation <> 'deleted' and payload.after.labels[0] = 'Question' THEN [1] ELSE [] END |

performance/docker-compose.yml

@@ -42,6 +42,7 @@ services:
       NEO4J_dbms_memory_heap_max__size: 2G
       NEO4J_kafka_max_poll_records: 16384
       NEO4J_streams_source_enabled: "false"
+      NEO4J_streams_sink_enabled: "true"
       NEO4J_streams_sink_topic_cypher_neo4j: "WITH event.payload AS payload, event.meta AS meta CALL apoc.do.case( [
       payload.type = 'node' AND meta.operation = 'created', \
       'CREATE (x:Performance {received_time: apoc.date.currentTimestamp()}) SET x+=props RETURN count(x)']