fixes #243: Restore the DLQ in Kafka Connect Sink (#244)

conker84 · jexp · commit 17c30967fd4f · 2019-09-18T07:36:09.000-07:00
diff --git a/common/src/main/kotlin/streams/extensions/CommonExtensions.kt b/common/src/main/kotlin/streams/extensions/CommonExtensions.kt
@@ -12,9 +12,15 @@ import org.neo4j.graphdb.Node
 import streams.serialization.JSONUtils
 import streams.service.StreamsSinkEntity
 import java.nio.ByteBuffer
+import java.util.*
 import javax.lang.model.SourceVersion
 
 fun Map<String,String>.getInt(name:String, defaultValue: Int) = this.get(name)?.toInt() ?: defaultValue
+fun Map<*, *>.asProperties() = this.let {
+    val properties = Properties()
+    properties.putAll(it)
+    properties
+}
 
 fun Node.labelNames() : List<String> {
     return this.labels.map { it.name() }
diff --git a/common/src/main/kotlin/streams/service/errors/KafkaErrorService.kt b/common/src/main/kotlin/streams/service/errors/KafkaErrorService.kt
@@ -1,28 +1,35 @@
 package streams.service.errors
 
 import org.apache.commons.lang3.exception.ExceptionUtils
+import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.Producer
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.clients.producer.ProducerRecord
 import org.apache.kafka.common.record.RecordBatch
 import org.apache.kafka.common.serialization.ByteArraySerializer
 import org.neo4j.util.VisibleForTesting
+import streams.utils.ValidationUtils.validateConnection
 import java.util.*
 
 class KafkaErrorService(private val producer: Producer<ByteArray, ByteArray>?, private val errorConfig: ErrorConfig, private val log: (String, Exception?)->Unit): ErrorService() {
 
     constructor(config: Properties, errorConfig: ErrorConfig,
-                log: (String, Exception?) -> Unit) : this(producer(errorConfig, config), errorConfig, log)
+                log: (String, Exception?) -> Unit) : this(producer(errorConfig, config, log), errorConfig, log)
 
     companion object {
-        private fun producer(errorConfig: ErrorConfig, config: Properties) =
+        private fun producer(errorConfig: ErrorConfig, config: Properties, log: (String, Exception?) -> Unit) =
                 errorConfig.dlqTopic?.let {
-                    val props = Properties()
-                    props.putAll(config)
-                    props[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java.name
-                    props[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java.name
-                    KafkaProducer<ByteArray,ByteArray>(props)
+                    try {
+                        val bootstrapServers = config.getOrDefault(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "").toString()
+                        validateConnection(bootstrapServers, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, false)
+                        config[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java.name
+                        config[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = ByteArraySerializer::class.java.name
+                        KafkaProducer<ByteArray, ByteArray>(config)
+                    } catch (e: Exception) {
+                        log("Cannot initialize the custom DLQ because of the following exception: ", e)
+                        null
+                    }
                 }
     }
 
@@ -35,8 +42,9 @@ class KafkaErrorService(private val producer: Producer<ByteArray, ByteArray>?, p
                 errorDatas.map { it.exception }.distinct().forEach{log("Error processing ${errorDatas.size} messages",it)}
             }
         }
-        if (errorConfig.dlqTopic != null && producer != null) {
-            errorDatas.forEach { dlqData ->
+
+        errorDatas.forEach { dlqData ->
+            producer?.let {
                 try {
                     val producerRecord = if (dlqData.timestamp == RecordBatch.NO_TIMESTAMP) {
                         ProducerRecord(errorConfig.dlqTopic, null, dlqData.key, dlqData.value)
@@ -47,9 +55,9 @@ class KafkaErrorService(private val producer: Producer<ByteArray, ByteArray>?, p
                         val producerHeader = producerRecord.headers()
                         populateContextHeaders(dlqData).forEach { (key, value) -> producerHeader.add(key, value) }
                     }
-                    producer.send(producerRecord)
+                    it.send(producerRecord)
                 } catch (e: Exception) {
-                    log("Error writing to DLQ $e :${dlqData.toLogString()}",e) // todo only the first or all
+                    log("Error writing to DLQ $e: ${dlqData.toLogString()}", e) // todo only the first or all
                 }
             }
         }
diff --git a/common/src/main/kotlin/streams/utils/ValidationUtils.kt b/common/src/main/kotlin/streams/utils/ValidationUtils.kt
@@ -32,7 +32,7 @@ object ValidationUtils {
         if (url.isBlank()) {
             throw RuntimeException("The `kafka.$kafkaPropertyKey` property is empty")
         } else if (checkReachable) {
-            val unreachableServers = ValidationUtils.checkServersUnreachable(url)
+            val unreachableServers = checkServersUnreachable(url)
             if (unreachableServers.isNotEmpty()) {
                 throw RuntimeException("The servers defined into the property `kafka.$kafkaPropertyKey` are not reachable: $unreachableServers")
             }
diff --git a/consumer/src/main/kotlin/streams/kafka/KafkaSinkConfiguration.kt b/consumer/src/main/kotlin/streams/kafka/KafkaSinkConfiguration.kt
@@ -79,7 +79,7 @@ data class KafkaSinkConfiguration(val zookeeperConnect: String = "localhost:2181
 
         private fun validate(config: KafkaSinkConfiguration) {
             validateConnection(config.zookeeperConnect, "zookeeper.connect", false)
-            validateConnection(config.bootstrapServers, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
+            validateConnection(config.bootstrapServers, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, false)
             val schemaRegistryUrlKey = "schema.registry.url"
             if (config.extraProperties.containsKey(schemaRegistryUrlKey)) {
                 val schemaRegistryUrl = config.extraProperties.getOrDefault(schemaRegistryUrlKey, "")
diff --git a/consumer/src/test/kotlin/streams/kafka/KafkaSinkConfigurationTest.kt b/consumer/src/test/kotlin/streams/kafka/KafkaSinkConfigurationTest.kt
@@ -4,6 +4,7 @@ import io.confluent.kafka.serializers.KafkaAvroDeserializer
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.apache.kafka.common.serialization.StringDeserializer
+import org.junit.Ignore
 import org.junit.Test
 import org.neo4j.kernel.configuration.Config
 import streams.StreamsSinkConfiguration
@@ -76,6 +77,7 @@ class KafkaSinkConfigurationTest {
     }
 
     @Test(expected = RuntimeException::class)
+    @Ignore("Disabled, use Kafka to deal with availability of the configured services")
     fun `should not validate the configuration because of unreachable kafka bootstrap server`() {
         val zookeeper = "zookeeper:2181"
         val bootstrap = "bootstrap:9092"
diff --git a/doc/asciidoc/consumer/index.adoc b/doc/asciidoc/consumer/index.adoc
@@ -483,7 +483,25 @@ Config Options
 | errors.deadletterqueue.topic.replication.factor | 3/1 | replication factor, need to set to 1 for single partition, default:3
 |===
 
-In Kafka Connect you can just configure the above properties.
+In Kafka Connect in addition to the above properties you need to define kafka broker connection properties:
+
+|===
+| Name | mandatory | Description
+
+| kafka.bootstrap.servers | true | It's the Kafka Broker url. *(please look at the description below)
+
+| kafka.<any_other_kafka_property> | false | You can also specify any other kafka Producer
+setting by adding the `kafka.` prefix (i.e the configuration `acks` become `kafka.acks`). See the https://kafka.apache.org/documentation/#brokerconfigs[Apache Kafka documentation] for details on these settings.
+
+|===
+
+*As you may have noticed we're asking to provide the `bootstrap.server` property,
+this because the Kafka Connect Framework provides an out-of-the-box support
+only for deserialization errors and message transformations
+(please look into the following link for further details: https://www.confluent.io/blog/kafka-connect-deep-dive-error-handling-dead-letter-queues).
+We want to extend this feature for transient errors in order to cover the 100% of failures.
+So to do that at this moment as suggested by Confluent we need to ask again the broker location,
+until this JIRA issue will not be addressed: https://issues.apache.org/jira/browse/KAFKA-8597.
 
 For the Neo4j extension you prefix them with `streams.sink.` in the Neo4j configuration.
 
diff --git a/kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jService.kt b/kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jService.kt
@@ -21,8 +21,6 @@ import streams.service.TopicType
 import streams.service.TopicTypeGroup
 import streams.utils.StreamsUtils
 import streams.utils.retryForException
-import java.lang.RuntimeException
-import java.util.concurrent.CompletionException
 import java.util.concurrent.CopyOnWriteArraySet
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeoutException
diff --git a/kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfig.kt b/kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfig.kt
@@ -10,19 +10,15 @@ import org.apache.kafka.common.config.AbstractConfig
 import org.apache.kafka.common.config.ConfigDef
 import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.connect.sink.SinkTask
-import org.neo4j.csv.reader.Magic.define
 import org.neo4j.driver.internal.async.pool.PoolSettings
 import org.neo4j.driver.v1.Config
-import streams.kafka.connect.sink.ConfigGroup.ERROR_REPORTING
 import streams.kafka.connect.utils.PropertiesUtil
 import streams.service.TopicType
 import streams.service.TopicUtils
 import streams.service.Topics
-import streams.service.errors.ErrorService
 import streams.service.sink.strategy.SourceIdIngestionStrategyConfig
 import java.io.File
 import java.net.URI
-import java.net.URISyntaxException
 import java.util.concurrent.TimeUnit
 
 enum class AuthenticationType {
@@ -71,6 +67,8 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
 
     val sourceIdStrategyConfig: SourceIdIngestionStrategyConfig
 
+    val kafkaBrokerProperties: Map<String, Any?>
+
     init {
         encryptionEnabled = getBoolean(ENCRYPTION_ENABLED)
         encryptionTrustStrategy = ConfigUtils
@@ -106,6 +104,10 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
         strategyMap = TopicUtils.toStrategyMap(topics, sourceIdStrategyConfig)
 
         parallelBatches = getBoolean(BATCH_PARALLELIZE)
+        val kafkaPrefix = "kafka."
+        kafkaBrokerProperties = originals
+                .filterKeys { it.toString().startsWith(kafkaPrefix) }
+                .mapKeys { it.key.toString().substring(kafkaPrefix.length) }
         validateAllTopics(originals)
     }
 
diff --git a/kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jSinkTask.kt b/kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jSinkTask.kt
@@ -7,7 +7,8 @@ import org.apache.kafka.connect.sink.SinkRecord
 import org.apache.kafka.connect.sink.SinkTask
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
-    import streams.service.errors.*
+import streams.extensions.asProperties
+import streams.service.errors.*
 import streams.utils.StreamsUtils
 import java.lang.Exception
 import java.util.*
@@ -17,38 +18,37 @@ class Neo4jSinkTask : SinkTask() {
     private val log: Logger = LoggerFactory.getLogger(Neo4jSinkTask::class.java)
     private lateinit var config: Neo4jSinkConnectorConfig
     private lateinit var neo4jService: Neo4jService
-//    private lateinit var errorService: ErrorService
+    private lateinit var errorService: ErrorService
 
     override fun version(): String {
         return VersionUtil.version(this.javaClass as Class<*>)
     }
 
     override fun start(map: Map<String, String>) {
         this.config = Neo4jSinkConnectorConfig(map)
-
-        val kafkaConfig = Properties()
-        kafkaConfig.putAll(map)
-//        this.errorService = KafkaErrorService(kafkaConfig, ErrorService.ErrorConfig.from(kafkaConfig), log::error)
         this.neo4jService = Neo4jService(this.config)
+        this.errorService = KafkaErrorService(this.config.kafkaBrokerProperties.asProperties(),
+                ErrorService.ErrorConfig.from(map.asProperties()),
+                log::error)
     }
 
     override fun put(collection: Collection<SinkRecord>) = runBlocking(Dispatchers.IO) {
         if (collection.isEmpty()) {
             return@runBlocking
         }
-//        try {
-        val data = EventBuilder()
-                .withBatchSize(config.batchSize)
-                .withTopics(config.topics.allTopics())
-                .withSinkRecords(collection)
-                .build()
-
-        neo4jService.writeData(data)
-//        } catch(e:Exception) {
-//            errorService.report(collection.map {
-//                ErrorData(it.topic(), it.timestamp(),it.key(), it.value(), it.kafkaPartition(), it.kafkaOffset(),  this::class.java,e)
-//            })
-//        }
+        try {
+            val data = EventBuilder()
+                    .withBatchSize(config.batchSize)
+                    .withTopics(config.topics.allTopics())
+                    .withSinkRecords(collection)
+                    .build()
+
+            neo4jService.writeData(data)
+        } catch(e:Exception) {
+            errorService.report(collection.map {
+                ErrorData(it.topic(), it.timestamp(),it.key(), it.value(), it.kafkaPartition(), it.kafkaOffset(),  this::class.java,e)
+            })
+        }
     }
 
     override fun stop() {
diff --git a/kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfigTest.kt b/kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfigTest.kt
@@ -1,5 +1,7 @@
 package streams.kafka.connect.sink
 
+import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.config.ConfigException
 import org.apache.kafka.connect.sink.SinkConnector
 import org.junit.Test
@@ -46,10 +48,14 @@ class Neo4jSinkConnectorConfigTest {
                 "${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}foo" to "CREATE (p:Person{name: event.firstName})",
                 Neo4jSinkConnectorConfig.SERVER_URI to "$a,$b", // Check for string trimming
                 Neo4jSinkConnectorConfig.BATCH_SIZE to 10,
+                "kafka.${CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG}" to "broker:9093",
+                "kafka.${ProducerConfig.ACKS_CONFIG}" to 1,
                 Neo4jSinkConnectorConfig.AUTHENTICATION_BASIC_USERNAME to "FOO",
                 Neo4jSinkConnectorConfig.AUTHENTICATION_BASIC_PASSWORD to "BAR")
         val config = Neo4jSinkConnectorConfig(originals)
 
+        assertEquals(mapOf(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG to "broker:9093",
+                ProducerConfig.ACKS_CONFIG to 1), config.kafkaBrokerProperties)
         assertEquals(originals["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}foo"], config.topics.cypherTopics["foo"])
         assertFalse { config.encryptionEnabled }
         assertEquals(a, config.serverUri.get(0).toString())
diff --git a/kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkTaskTest.kt b/kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkTaskTest.kt
@@ -128,7 +128,6 @@ class Neo4jSinkTaskTest {
         val firstTopic = "neotopic"
         val secondTopic = "foo"
         val props = mutableMapOf<String, String>()
-        props[Neo4jSinkConnectorConfig.ENCRYPTION_ENABLED] = false.toString()
         props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
         props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$firstTopic"] = "CREATE (n:PersonExt {name: event.firstName, surname: event.lastName})"
         props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$secondTopic"] = "CREATE (n:Person {name: event.firstName})"
@@ -367,7 +366,6 @@ class Neo4jSinkTaskTest {
     fun `should not insert data into Neo4j`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()
-        props[Neo4jSinkConnectorConfig.ENCRYPTION_ENABLED] = false.toString()
         props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
         props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$topic"] = "CREATE (n:Person {name: event.firstName, surname: event.lastName})"
         props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
@@ -389,11 +387,9 @@ class Neo4jSinkTaskTest {
     }
 
     @Test
-    @Ignore("temporary disabled the DLQ")
     fun `should report but not fail parsing data`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()
-        props[Neo4jSinkConnectorConfig.ENCRYPTION_ENABLED] = false.toString()
         props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
         props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$topic"] = "CREATE (n:Person {name: event.firstName, surname: event.lastName})"
         props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
@@ -412,11 +408,9 @@ class Neo4jSinkTaskTest {
     }
 
     @Test
-    @Ignore("temporary disabled the DLQ")
     fun `should report but not fail invalid schema`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()
-        props[Neo4jSinkConnectorConfig.ENCRYPTION_ENABLED] = false.toString()
         props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
         props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$topic"] = "CREATE (n:Person {name: event.firstName, surname: event.lastName})"
         props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
@@ -435,11 +429,9 @@ class Neo4jSinkTaskTest {
     }
 
     @Test
-    @Ignore("temporary disabled the DLQ")
     fun `should fail running invalid cypher`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()
-        props[Neo4jSinkConnectorConfig.ENCRYPTION_ENABLED] = false.toString()
         props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
         props["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}$topic"] = " No Valid Cypher "
         props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
diff --git a/producer/src/main/kotlin/streams/StreamsExtensionFactory.kt b/producer/src/main/kotlin/streams/StreamsExtensionFactory.kt
@@ -11,6 +11,7 @@ import org.neo4j.kernel.lifecycle.Lifecycle
 import org.neo4j.kernel.lifecycle.LifecycleAdapter
 import org.neo4j.logging.internal.LogService
 import streams.procedures.StreamsProcedures
+import streams.utils.StreamsUtils
 
 class StreamsExtensionFactory : KernelExtensionFactory<StreamsExtensionFactory.Dependencies>(ExtensionType.DATABASE,"Streams.Producer") {
     override fun newInstance(context: KernelContext, dependencies: Dependencies): Lifecycle {
@@ -69,17 +70,13 @@ class StreamsEventRouterLifecycle(val db: GraphDatabaseAPI, val configuration: C
 
     private fun unregisterTransactionEventHandler() {
         if (streamsEventRouterConfiguration.enabled) {
-            if (::streamsConstraintsService.isInitialized) {
-                streamsConstraintsService.close()
-            }
+            StreamsUtils.ignoreExceptions({ streamsConstraintsService.close() }, UninitializedPropertyAccessException::class.java)
             db.unregisterTransactionEventHandler(txHandler)
         }
     }
 
     override fun stop() {
         unregisterTransactionEventHandler()
-        if (::streamHandler.isInitialized) {
-            streamHandler.stop()
-        }
+        StreamsUtils.ignoreExceptions({ streamHandler.stop() }, UninitializedPropertyAccessException::class.java)
     }
 }
diff --git a/producer/src/main/kotlin/streams/kafka/KafkaConfiguration.kt b/producer/src/main/kotlin/streams/kafka/KafkaConfiguration.kt
@@ -63,7 +63,7 @@ data class KafkaConfiguration(val zookeeperConnect: String = "localhost:2181",
 
         private fun validate(config: KafkaConfiguration) {
             validateConnection(config.zookeeperConnect, "zookeeper.connect", false)
-            validateConnection(config.bootstrapServers, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
+            validateConnection(config.bootstrapServers, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, false)
         }
 
     }
diff --git a/producer/src/main/kotlin/streams/kafka/KafkaEventRouter.kt b/producer/src/main/kotlin/streams/kafka/KafkaEventRouter.kt
@@ -15,6 +15,7 @@ import streams.StreamsEventRouter
 import streams.events.StreamsEvent
 import streams.events.StreamsTransactionEvent
 import streams.serialization.JSONUtils
+import streams.utils.StreamsUtils
 import java.util.*
 import java.util.concurrent.ThreadLocalRandom
 
@@ -39,9 +40,7 @@ class KafkaEventRouter: StreamsEventRouter {
     }
 
     override fun stop() {
-        if (::producer.isInitialized) {
-            producer.close()
-        }
+        StreamsUtils.ignoreExceptions({ producer.close() }, UninitializedPropertyAccessException::class.java)
     }
 
     private fun send(producerRecord: ProducerRecord<String, ByteArray>) {

Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,7 @@ object ValidationUtils {`
`32`	`32`	`if (url.isBlank()) {`
`33`	`33`	throw RuntimeException("The `kafka.$kafkaPropertyKey` property is empty")
`34`	`34`	`} else if (checkReachable) {`
`35`		`- val unreachableServers = ValidationUtils.checkServersUnreachable(url)`
	`35`	`+ val unreachableServers = checkServersUnreachable(url)`
`36`	`36`	`if (unreachableServers.isNotEmpty()) {`
`37`	`37`	throw RuntimeException("The servers defined into the property `kafka.$kafkaPropertyKey` are not reachable: $unreachableServers")
`38`	`38`	`}`