Fixes #464: Kafka sink ignores neo4j.batch.timeout.msecs when neo4j.batch.parallelize = false (#468)

Author: vga91

doc/docs/modules/ROOT/pages/architecture/throughput.adoc

@@ -17,9 +17,12 @@ And so a critical factor is how many items are in the "events" array, per-transa
 
 * Heap sizing affects how big transactions coming from Kafka can be without starving other queries that are running
 * Page cache sizing affects how much data is hot and has a major impact on the cypher queries that neo4j-streams runs.
-* `neo4j.batch.parallelize` affects whether batches can be run in parallel.  This is a potential massive benefit for throughput - but comes with a sharp edge that ordering cannot be guaranteed if you parallelize, and there is risk of locking errors.    **This setting only applies to the kafka connect worker**.
-* `neo4j.batch.size` affects how many records get run in a Neo4j batch.
-* `neo4j.batch.timeout.msecs` affects how long batches are given to execute.
+* `neo4j.batch.parallelize` affects whether batches can be run in parallel.  This is a potential massive benefit for throughput - but comes with a sharp edge that ordering cannot be guaranteed if you parallelize, and there is risk of locking errors. (`*`)
+* `neo4j.batch.size` affects how many records get run in a Neo4j batch. (`*`)
+* `neo4j.batch.timeout.msecs` affects how long batches are given to execute. (`*`)
+
+**The settings with `*` only applies to the kafka connect worker**.
+
 
 == Bolt Protocol overhead & Latency
 

doc/docs/modules/ROOT/pages/kafka-connect.adoc

@@ -206,7 +206,7 @@ include::cud-file-format.adoc[]
 
 Neo4j 4.0 Enterprise has https://neo4j.com/docs/operations-manual/4.0/manage-databases/[multi-tenancy support],
 in order to support this feature, in order to support this feature with Kafka Connect plugin, creating the Sink
-instance we have to add the `database` property, which tells the Connector the database to use as default. If you
+instance we have to add the `neo4j.database` property, which tells the Connector the database to use as default. If you
 don't specify that property, the default database `neo4j` will be used.
 
 [NOTE]
@@ -303,7 +303,6 @@ Following a summary of all the configuration parameters you can use for the Kafk
 | Mandatory
 | Note
 
-| database | <DATABASE_NAME> | false | Specify a database name only if you want to use a non-default database. Default value is 'neo4j'
 | topics | <topicA,topicB> | true | A list of comma-separated topics
 | connector.class | streams.kafka.connect.sink.Neo4jSinkConnector | true |
 | key.converter | org.apache.kafka.connect.storage.StringConverter | false | Converter class for key Connect data
@@ -321,10 +320,25 @@ Following a summary of all the configuration parameters you can use for the Kafk
 | errors.deadletterqueue.context.headers.enable | false/true | false | enrich messages with metadata headers like exception, timestamp, org. topic, org.part, default:false
 | errors.deadletterqueue.context.headers.prefix | prefix-text | false | common prefix for header entries, e.g. `"__streams.errors."` , default: not set
 | errors.deadletterqueue.topic.replication.factor | 3/1 | false | replication factor, need to set to 1 for single partition, default:3
+| neo4j.database | "bolt://neo4j:7687" | true | Specify a database name only if you want to use a non-default database. Default value is 'neo4j'
 | neo4j.server.uri | "bolt://neo4j:7687" | true | Neo4j Server URI
 | neo4j.authentication.basic.username | your_neo4j_user | true | Neo4j username
 | neo4j.authentication.basic.password | your_neo4j_password | true | Neo4j password
+| neo4j.authentication.basic.realm | your_neo4j_auth_realm | false | The authentication realm
+| neo4j.authentication.kerberos.ticket | your_kerberos_ticket | false | The Kerberos ticket
+| neo4j.authentication.type | NONE/BASIC/KERBEROS | false | The authentication type (default: 'BASIC')
+| neo4j.batch.size | Integer | false | The max number of events processed by the Cypher query (default: 1000)
+| neo4j.batch.timeout.msecs | Integer | false | The execution timeout for the cypher query (default: 0, that is without timeout)
+| neo4j.batch.parallelize | boolean | false | If enabled messages are processed concurrently in the sink. Non concurrent execution supports in-order processing, e.g. for CDC
+| neo4j.connection.max.lifetime.msecs | Long | false | The max Neo4j connection lifetime (default: 1 hour)
+| neo4j.connection.acquisition.timeout.msecs | Long | false | The max Neo4j acquisition timeout (default 1 hour)
+| neo4j.connection.liveness.check.timeout.msecs | Long | false | The max Neo4j liveness check timeout (default 1 hour)
+| neo4j.connection.max.pool.size | Int | false | The max pool size (default: 100)
+| neo4j.encryption.ca.certificate.path | your_certificate_path | false | The path of the certificate
 | neo4j.encryption.enabled | true/false | false |
+| neo4j.encryption.trust.strategy | TRUST_ALL_CERTIFICATES/TRUST_CUSTOM_CA_SIGNED_CERTIFICATES/TRUST_SYSTEM_CA_SIGNED_CERTIFICATES | false | The Neo4j trust strategy (default: TRUST_ALL_CERTIFICATES)
+| neo4j.retry.backoff.msecs | Long | false | The time in milliseconds to wait following a transient error before a retry attempt is made (default: 30000).
+| neo4j.retry.max.attemps | Long | false | The maximum number of times to retry on transient errors (except for TimeoutException) before failing the task (default: 5).
 | neo4j.topic.cdc.sourceId | <list of topics separated by semicolon> | false |
 | neo4j.topic.cdc.sourceId.labelName | <the label attached to the node> | false | default value is *SourceEvent*
 | neo4j.topic.cdc.sourceId.idName | <the id name given to the CDC id field> | false | default value is *sourceId*

doc/docs/modules/ROOT/partials/docker-data/kafka-connect-docker-compose.yml

@@ -96,7 +96,7 @@ services:
       CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
       CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
       CONNECT_ZOOKEEPER_CONNECT: 'zookeeper:2181'
-      CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components
+      CONNECT_PLUGIN_PATH: /usr/share/java,/tmp/connect-plugins
       CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=DEBUG,org.I0Itec.zkclient=DEBUG,org.reflections=ERROR
     command: 
 #      - bash 

kafka-connect-neo4j/docker/docker-compose.yml

@@ -96,7 +96,7 @@ services:
       CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
       CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
       CONNECT_ZOOKEEPER_CONNECT: 'zookeeper:2181'
-      CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components
+      CONNECT_PLUGIN_PATH: /usr/share/java,/tmp/connect-plugins
       CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=DEBUG,org.I0Itec.zkclient=DEBUG,org.reflections=ERROR
     command: 
 #      - bash 

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

@@ -4,6 +4,7 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.ObsoleteCoroutinesApi
 import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.runBlocking
 import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.connect.errors.ConnectException
@@ -12,17 +13,18 @@ import org.neo4j.driver.Config
 import org.neo4j.driver.Driver
 import org.neo4j.driver.GraphDatabase
 import org.neo4j.driver.SessionConfig
+import org.neo4j.driver.TransactionConfig
 import org.neo4j.driver.exceptions.ClientException
 import org.neo4j.driver.exceptions.TransientException
 import org.neo4j.driver.net.ServerAddress
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
-import streams.extensions.awaitAll
 import streams.extensions.errors
 import streams.kafka.connect.utils.PropertiesUtil
 import streams.service.StreamsSinkEntity
 import streams.service.StreamsSinkService
 import streams.utils.retryForException
+import java.time.Duration
 import java.util.concurrent.TimeUnit
 import kotlin.streams.toList
 
@@ -34,6 +36,7 @@ class Neo4jService(private val config: Neo4jSinkConnectorConfig):
 
     private val driver: Driver
     private val sessionConfig: SessionConfig
+    private val transactionConfig: TransactionConfig
 
     init {
         val configBuilder = Config.builder()
@@ -80,6 +83,15 @@ class Neo4jService(private val config: Neo4jSinkConnectorConfig):
             sessionConfigBuilder.withDatabase(config.database)
         }
         this.sessionConfig = sessionConfigBuilder.build()
+        
+        val batchTimeout = this.config.batchTimeout
+        this.transactionConfig = if (batchTimeout > 0) {
+            TransactionConfig.builder()
+                    .withTimeout(Duration.ofMillis(batchTimeout))
+                    .build()
+        } else {
+            TransactionConfig.empty()
+        }
     }
 
     fun close() {
@@ -91,14 +103,15 @@ class Neo4jService(private val config: Neo4jSinkConnectorConfig):
         driver.session(sessionConfig).use { session ->
             try {
                 runBlocking {
-                    retryForException<Unit>(exceptions = arrayOf(ClientException::class.java, TransientException::class.java),
-                            retries = config.retryMaxAttempts, delayTime = 0) { // we use the delayTime = 0, because we delegate the retryBackoff to the Neo4j Java Driver
-                        session.writeTransaction {
+                    retryForException(exceptions = arrayOf(ClientException::class.java, TransientException::class.java),
+                            retries = config.retryMaxAttempts, delayTime = 0) {  // we use the delayTime = 0, because we delegate the retryBackoff to the Neo4j Java Driver
+
+                        session.writeTransaction({
                             val summary = it.run(query, data).consume()
                             if (log.isDebugEnabled) {
                                 log.debug("Successfully executed query: `$query`. Summary: $summary")
                             }
-                        }
+                        }, transactionConfig)
                     }
                 }
             } catch (e: Exception) {
@@ -129,7 +142,8 @@ class Neo4jService(private val config: Neo4jSinkConnectorConfig):
                     records.map { async(Dispatchers.IO) { writeForTopic(topic, it) } }
                 }
 
-        jobs.awaitAll(config.batchTimeout)
+        // timeout starts in writeTransaction()
+        jobs.awaitAll()
         jobs.mapNotNull { it.errors() }
     }
 

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

@@ -169,7 +169,7 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
         const val TOPIC_CUD = "neo4j.topic.cud"
 
         const val CONNECTION_POOL_MAX_SIZE_DEFAULT = 100
-        val BATCH_TIMEOUT_DEFAULT = TimeUnit.SECONDS.toMillis(30L)
+        val BATCH_TIMEOUT_DEFAULT = TimeUnit.SECONDS.toMillis(0L)
         const val BATCH_SIZE_DEFAULT = 1000
         val RETRY_BACKOFF_DEFAULT = TimeUnit.SECONDS.toMillis(30L)
         const val RETRY_MAX_ATTEMPTS_DEFAULT = 5
@@ -298,7 +298,7 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
                             .importance(ConfigDef.Importance.LOW)
                             .defaultValue(BATCH_TIMEOUT_DEFAULT)
                             .group(ConfigGroup.BATCH)
-                            .validator(ConfigDef.Range.atLeast(1)).build())
+                            .validator(ConfigDef.Range.atLeast(0)).build())
                     .define(ConfigKeyBuilder
                             .of(RETRY_BACKOFF_MSECS, ConfigDef.Type.LONG)
                             .documentation(PropertiesUtil.getProperty(RETRY_BACKOFF_MSECS))

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

@@ -17,21 +17,21 @@ neo4j.database=Type: String;\nDescription: The neo4j database instance name (def
 neo4j.server.uri=Type: String;\nDescription: The Bolt URI (default bolt://localhost:7687)
 neo4j.authentication.type=Type: enum[NONE, BASIC, KERBEROS];\nDescription: The authentication type (default BASIC)
 neo4j.batch.size=Type: Int;\nDescription: The max number of events processed by the Cypher query (default 1000)
-neo4j.batch.timeout.msecs=Type: Long;\nDescription: The execution timeout for the cypher query (default 30000)
+neo4j.batch.timeout.msecs=Type: Long;\nDescription: The execution timeout for the cypher query (default: 0, that is without timeout)
 neo4j.authentication.basic.username=Type: String;\nDescription: The authentication username
 neo4j.authentication.basic.password=Type: String;\nDescription: The authentication password
 neo4j.authentication.basic.realm=Type: String;\nDescription: The authentication realm
 neo4j.authentication.kerberos.ticket=Type: String;\nDescription: The Kerberos ticket
 neo4j.encryption.enabled=Type: Boolean;\nDescription: If the encryption is enabled (default false)
-neo4j.encryption.trust.strategy=Type: enum[TRUST_ALL_CERTIFICATES, TRUST_CUSTOM_CA_SIGNED_CERTIFICATES, TRUST_SYSTEM_CA_SIGNED_CERTIFICATES];\nDescription: The Neo4j trust strategy (default RUST_ALL_CERTIFICATES)
+neo4j.encryption.trust.strategy=Type: enum[TRUST_ALL_CERTIFICATES, TRUST_CUSTOM_CA_SIGNED_CERTIFICATES, TRUST_SYSTEM_CA_SIGNED_CERTIFICATES];\nDescription: The Neo4j trust strategy (default TRUST_ALL_CERTIFICATES)
 neo4j.encryption.ca.certificate.path=Type: String;\nDescription: The path of the certificate
 neo4j.connection.max.lifetime.msecs=Type: Long;\nDescription: The max Neo4j connection lifetime (default 1 hour)
 neo4j.connection.acquisition.timeout.msecs=Type: Long;\nDescription: The max Neo4j acquisition timeout (default 1 hour)
 neo4j.connection.liveness.check.timeout.msecs=Type: Long;\nDescription: The max Neo4j liveness check timeout (default 1 hour)
 neo4j.connection.max.pool.size=Type: Int;\nDescription: The max pool size (default 100)
 neo4j.load.balance.strategy=Type: enum[ROUND_ROBIN, LEAST_CONNECTED];\nDescription: The Neo4j load balance strategy (default LEAST_CONNECTED)
 neo4j.retry.backoff.msecs=Type: Long;\nDescription: The time in milliseconds to wait following a transient error before a retry attempt is made (default 30000).
-neo4j.retry.max.attemps=Type: Int;\nDescription: The maximum number of times to retry on transient errors before failing the task (default 5).
+neo4j.retry.max.attemps=Type: Int;\nDescription: The maximum number of times to retry on transient errors (except for TimeoutException) before failing the task (default 5).
 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)

kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkTaskTest.kt

@@ -7,6 +7,7 @@ import org.apache.kafka.connect.data.Timestamp
 import org.apache.kafka.connect.sink.SinkRecord
 import org.apache.kafka.connect.sink.SinkTask
 import org.apache.kafka.connect.sink.SinkTaskContext
+import org.junit.After
 import org.junit.Rule
 import org.junit.Test
 import org.mockito.Mockito.mock
@@ -38,6 +39,11 @@ class Neo4jSinkTaskTest {
             .withDisabledServer()
             .withConfig(GraphDatabaseSettings.auth_enabled, false)
 
+    @After
+    fun after() {
+        cleanAll()
+    }
+
     private val PERSON_SCHEMA = SchemaBuilder.struct().name("com.example.Person")
             .field("firstName", Schema.STRING_SCHEMA)
             .field("lastName", Schema.STRING_SCHEMA)
@@ -1276,5 +1282,72 @@ class Neo4jSinkTaskTest {
                     && errorData.exception!!.javaClass.name == "org.neo4j.driver.exceptions.ClientException")
         }
     }
+    
+    @Test
+    fun `should stop the query and fails with small timeout and vice versa`() {
+        val myTopic = "foo"
+        val props = mutableMapOf<String, String>()
+        props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
+        props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$myTopic"] = "CREATE (n:Person {name: event.name})"
+        props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
+        props[Neo4jSinkConnectorConfig.BATCH_PARALLELIZE] = true.toString()
+        val batchSize = 500000
+        props[Neo4jSinkConnectorConfig.BATCH_SIZE] = batchSize.toString()
+        props[Neo4jSinkConnectorConfig.BATCH_TIMEOUT_MSECS] = 1.toString()
+        props[SinkTask.TOPICS_CONFIG] = myTopic
+        val input = (1..batchSize).map {
+            SinkRecord(myTopic, 1, null, null, null, mapOf("name" to it.toString()), it.toLong())
+        }
+        // test timeout with parallel=true
+        assertFailsWithTimeout(props, input, batchSize)
+        countFooPersonEntities(0)
+
+        // test timeout with parallel=false
+        props[Neo4jSinkConnectorConfig.BATCH_PARALLELIZE] = false.toString()
+        assertFailsWithTimeout(props, input, batchSize)
+        countFooPersonEntities(0)
+
+        // test with large timeout
+        props[Neo4jSinkConnectorConfig.BATCH_TIMEOUT_MSECS] = 30000.toString()
+        val taskValidParallelFalse = Neo4jSinkTask()
+        taskValidParallelFalse.initialize(mock(SinkTaskContext::class.java))
+        taskValidParallelFalse.start(props)
+        taskValidParallelFalse.put(input)         
+        countFooPersonEntities(batchSize)
+
+        props[Neo4jSinkConnectorConfig.BATCH_PARALLELIZE] = true.toString()
+        val taskValidParallelTrue = Neo4jSinkTask()
+        taskValidParallelTrue.initialize(mock(SinkTaskContext::class.java))
+        taskValidParallelTrue.start(props)
+        taskValidParallelTrue.put(input)         
+        countFooPersonEntities(batchSize * 2)
+    }
+
+    private fun assertFailsWithTimeout(props: MutableMap<String, String>, input: List<SinkRecord>, expectedDataErrorSize: Int) {
+        try {
+            val taskInvalid = Neo4jSinkTask()
+            taskInvalid.initialize(mock(SinkTaskContext::class.java))
+            taskInvalid.start(props)
+            taskInvalid.put(input)
+            fail("Should fail because of TimeoutException")
+        } catch (e: ProcessingError) {
+            val errors = e.errorDatas
+            assertEquals(expectedDataErrorSize, errors.size)
+        }
+    }
+
+    private fun countFooPersonEntities(expected: Int) {
+        db.defaultDatabaseService().beginTx().use {
+            val personCount = it.execute("MATCH (p:Person) RETURN count(p) as count").columnAs<Long>("count").next()
+            assertEquals(expected, personCount.toInt())
+        }
+    }
+    
+    private fun cleanAll() {
+        db.defaultDatabaseService().beginTx().use {
+            it.execute("MATCH (n) DETACH DELETE n")
+            it.commit()
+        }
+    }
 
 }