Merge pull request #204 from conker84/issue_177

mneedham · web-flow · commit 4b9636218bdb · 2019-06-25T14:09:01.000+01:00
fixes #177: Create a Dead Letter Queue
diff --git a/common/src/main/kotlin/streams/service/dlq/DeadLetterQueueService.kt b/common/src/main/kotlin/streams/service/dlq/DeadLetterQueueService.kt
@@ -0,0 +1,43 @@
+package streams.service.dlq
+
+import org.apache.kafka.clients.consumer.ConsumerRecord
+
+
+data class DLQData(val originalTopic: String,
+                   val timestamp: Long,
+                   val key: ByteArray,
+                   val value: ByteArray,
+                   val partition: String,
+                   val offset: String,
+                   val executingClass: Class<*>?,
+                   val exception: Exception?) {
+
+    companion object {
+        fun from(consumerRecord: ConsumerRecord<String, ByteArray>, exception: Exception?, executingClass: Class<*>?): DLQData {
+            return DLQData(offset = consumerRecord.offset().toString(),
+                    originalTopic = consumerRecord.topic(),
+                    partition = consumerRecord.partition().toString(),
+                    timestamp = consumerRecord.timestamp(),
+                    exception = exception,
+                    executingClass = executingClass,
+                    key = consumerRecord.key().toByteArray(),
+                    value = consumerRecord.value())
+        }
+    }
+}
+
+abstract class DeadLetterQueueService(private val config: Map<String, Any>,
+                                             headerPrefix: String) {
+
+    val ERROR_HEADER_ORIG_TOPIC = headerPrefix + "topic"
+    val ERROR_HEADER_ORIG_PARTITION = headerPrefix + "partition"
+    val ERROR_HEADER_ORIG_OFFSET = headerPrefix + "offset"
+    val ERROR_HEADER_EXECUTING_CLASS = headerPrefix + "class.name"
+    val ERROR_HEADER_EXCEPTION = headerPrefix + "exception.class.name"
+    val ERROR_HEADER_EXCEPTION_MESSAGE = headerPrefix + "exception.message"
+    val ERROR_HEADER_EXCEPTION_STACK_TRACE = headerPrefix + "exception.stacktrace"
+
+    abstract fun send(deadLetterQueueTopic: String, dlqData: DLQData)
+
+    abstract fun close()
+}
diff --git a/common/src/main/kotlin/streams/service/dlq/KafkaDLQService.kt b/common/src/main/kotlin/streams/service/dlq/KafkaDLQService.kt
@@ -0,0 +1,67 @@
+package streams.service.dlq
+
+import org.apache.commons.lang3.exception.ExceptionUtils
+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 java.util.*
+
+class KafkaDLQService: DeadLetterQueueService {
+
+    private var producer: Producer<ByteArray, ByteArray>
+
+    constructor(config: Map<String, Any>,
+                headerPrefix: String): super(config, headerPrefix) { // "__connect.errors."
+        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
+        producer = KafkaProducer(props)
+    }
+
+    @VisibleForTesting
+    constructor(producer: Producer<ByteArray, ByteArray>, headerPrefix: String = ""): super(emptyMap(), headerPrefix) {
+        this.producer = producer
+    }
+
+    override fun send(deadLetterQueueTopic: String, dlqData: DLQData) {
+        val producerRecord = if (dlqData.timestamp == RecordBatch.NO_TIMESTAMP) {
+            ProducerRecord(deadLetterQueueTopic, null,
+                    dlqData.key, dlqData.value)
+        } else {
+            ProducerRecord(deadLetterQueueTopic, null, dlqData.timestamp,
+                    dlqData.key, dlqData.value)
+        }
+        val producerHeader = producerRecord.headers()
+        populateContextHeaders(dlqData).forEach { key, value -> producerHeader.add(key, value) }
+        producer.send(producerRecord)
+    }
+
+    @VisibleForTesting
+    fun populateContextHeaders(dlqData: DLQData): Map<String, ByteArray> {
+        val headers = mutableMapOf<String, ByteArray>()
+        headers[ERROR_HEADER_ORIG_TOPIC] = dlqData.originalTopic.toByteArray()
+        headers[ERROR_HEADER_ORIG_PARTITION] = dlqData.partition.toByteArray()
+        headers[ERROR_HEADER_ORIG_OFFSET] = dlqData.offset.toByteArray()
+        if (dlqData.executingClass != null) {
+            headers[ERROR_HEADER_EXECUTING_CLASS] = dlqData.executingClass.name.toByteArray()
+        }
+        if (dlqData.exception != null) {
+            headers[ERROR_HEADER_EXCEPTION] = dlqData.exception.javaClass.name.toByteArray()
+            if (dlqData.exception.message != null) {
+                headers[ERROR_HEADER_EXCEPTION_MESSAGE] = dlqData.exception.message.toString().toByteArray()
+            }
+            headers[ERROR_HEADER_EXCEPTION_STACK_TRACE] = ExceptionUtils.getStackTrace(dlqData.exception).toByteArray()
+        }
+        return headers
+    }
+
+    override fun close() {
+        this.producer.close()
+    }
+
+}
diff --git a/common/src/test/kotlin/streams/service/sink/dlq/KafkaDLQServiceTest.kt b/common/src/test/kotlin/streams/service/sink/dlq/KafkaDLQServiceTest.kt
@@ -0,0 +1,78 @@
+package streams.service.sink.dlq
+
+import org.apache.commons.lang3.exception.ExceptionUtils
+import org.apache.kafka.clients.producer.MockProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.internals.FutureRecordMetadata
+import org.apache.kafka.common.record.RecordBatch
+import org.junit.Test
+import org.mockito.ArgumentMatchers
+import org.mockito.Mockito
+import streams.service.dlq.DLQData
+import streams.service.dlq.KafkaDLQService
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.assertEquals
+
+class KafkaDLQServiceTest {
+    @Test
+    fun `should send the data to the DLQ`() {
+        val producer: MockProducer<ByteArray, ByteArray> = Mockito.mock(MockProducer::class.java) as MockProducer<ByteArray, ByteArray>
+        val counter = AtomicInteger(0)
+        Mockito.`when`(producer.send(ArgumentMatchers.any<ProducerRecord<ByteArray, ByteArray>>())).then {
+            counter.incrementAndGet()
+            FutureRecordMetadata(null, 0, RecordBatch.NO_TIMESTAMP, 0L, 0, 0)
+        }
+        val dlqService = KafkaDLQService(producer)
+        val offset = "0"
+        val originalTopic = "topicName"
+        val partition = "1"
+        val timestamp = System.currentTimeMillis()
+        val exception = RuntimeException("Test")
+        val key = "KEY"
+        val value = "VALUE"
+        val dlqData = DLQData(offset = offset,
+                originalTopic = originalTopic,
+                partition = partition,
+                timestamp = timestamp,
+                exception = exception,
+                executingClass = KafkaDLQServiceTest::class.java,
+                key = key.toByteArray(),
+                value = value.toByteArray())
+        dlqService.send("dlqTopic", dlqData)
+        assertEquals(1, counter.get())
+        dlqService.close()
+    }
+
+
+    @Test
+    fun `should create the header map`() {
+        val producer: MockProducer<ByteArray, ByteArray> = Mockito.mock(MockProducer::class.java) as MockProducer<ByteArray, ByteArray>
+        val dlqService = KafkaDLQService(producer)
+        val offset = "0"
+        val originalTopic = "topicName"
+        val partition = "1"
+        val timestamp = System.currentTimeMillis()
+        val exception = RuntimeException("Test")
+        val key = "KEY"
+        val value = "VALUE"
+        val dlqData = DLQData(
+                offset = offset,
+                originalTopic = originalTopic,
+                partition = partition,
+                timestamp = timestamp,
+                exception = exception,
+                executingClass = KafkaDLQServiceTest::class.java,
+                key = key.toByteArray(),
+                value = value.toByteArray()
+        )
+        val map = dlqService.populateContextHeaders(dlqData)
+        assertEquals(String(map["topic"]!!), originalTopic)
+        assertEquals(String(map["partition"]!!), partition)
+        assertEquals(String(map["offset"]!!), offset)
+        assertEquals(String(map["class.name"]!!), KafkaDLQServiceTest::class.java.name)
+        assertEquals(String(map["exception.class.name"]!!), exception::class.java.name)
+        assertEquals(String(map["exception.message"]!!), exception.message)
+        assertEquals(String(map["exception.stacktrace"]!!), ExceptionUtils.getStackTrace(exception))
+
+    }
+}
diff --git a/consumer/src/main/kotlin/streams/StreamsEventSink.kt b/consumer/src/main/kotlin/streams/StreamsEventSink.kt
@@ -3,6 +3,7 @@ package streams
 import org.neo4j.kernel.configuration.Config
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.logging.Log
+import streams.service.dlq.DeadLetterQueueService
 
 abstract class StreamsEventSink(private val config: Config,
                                 private val queryExecution: StreamsEventSinkQueryExecution,
@@ -20,7 +21,7 @@ abstract class StreamsEventSink(private val config: Config,
 
 }
 
-abstract class StreamsEventConsumer(private val log: Log) {
+abstract class StreamsEventConsumer(private val log: Log, private val dlqService: DeadLetterQueueService?) {
 
     abstract fun stop()
 
diff --git a/consumer/src/main/kotlin/streams/StreamsSinkConfiguration.kt b/consumer/src/main/kotlin/streams/StreamsSinkConfiguration.kt
@@ -16,6 +16,7 @@ data class StreamsSinkConfiguration(val enabled: Boolean = false,
                                     val proceduresEnabled: Boolean = true,
                                     val sinkPollingInterval: Long = 10000,
                                     val topics: Topics = Topics(),
+                                    val dlqTopic: String = "",
                                     val sourceIdStrategyConfig: SourceIdIngestionStrategyConfig = SourceIdIngestionStrategyConfig()) {
 
     companion object {
@@ -39,11 +40,14 @@ data class StreamsSinkConfiguration(val enabled: Boolean = false,
                     config.getOrDefault("sink.topic.cdc.sourceId.labelName", defaultSourceIdStrategyConfig.labelName),
                     config.getOrDefault("sink.topic.cdc.sourceId.idName", defaultSourceIdStrategyConfig.idName))
 
+            val dlqTopic = config.getOrDefault("sink.dlq", "")
+
             return default.copy(enabled = config.getOrDefault(StreamsSinkConfigurationConstants.ENABLED, default.enabled).toString().toBoolean(),
                     proceduresEnabled = config.getOrDefault(StreamsSinkConfigurationConstants.PROCEDURES_ENABLED, default.proceduresEnabled)
                             .toString().toBoolean(),
                     sinkPollingInterval = config.getOrDefault("sink.polling.interval", default.sinkPollingInterval).toString().toLong(),
                     topics = topics,
+                    dlqTopic = dlqTopic,
                     sourceIdStrategyConfig = sourceIdStrategyConfig)
         }
 
diff --git a/consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt b/consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt
@@ -1,10 +1,7 @@
 package streams.kafka
 
 import kotlinx.coroutines.*
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.apache.kafka.clients.consumer.KafkaConsumer
-import org.apache.kafka.clients.consumer.OffsetAndMetadata
+import org.apache.kafka.clients.consumer.*
 import org.apache.kafka.common.TopicPartition
 import org.neo4j.kernel.configuration.Config
 import org.neo4j.kernel.internal.GraphDatabaseAPI
@@ -13,7 +10,11 @@ import streams.*
 import streams.extensions.offsetAndMetadata
 import streams.extensions.topicPartition
 import streams.serialization.JSONUtils
+import streams.service.dlq.DLQData
+import streams.service.dlq.KafkaDLQService
 import streams.utils.Neo4jUtils
+import streams.utils.StreamsUtils
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.TimeUnit
 
 
@@ -41,10 +42,20 @@ class KafkaEventSink(private val config: Config,
         return object: StreamsEventConsumerFactory() {
             override fun createStreamsEventConsumer(config: Map<String, String>, log: Log): StreamsEventConsumer {
                 val kafkaConfig = KafkaSinkConfiguration.from(config)
+                val dlqService = if (kafkaConfig.streamsSinkConfiguration.dlqTopic.isNotBlank()) {
+                    val asProperties = kafkaConfig.asProperties()
+                            .mapKeys { it.key.toString() }
+                            .toMutableMap()
+                    asProperties.remove(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+                    asProperties.remove(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+                    KafkaDLQService(asProperties, "__streams.errors")
+                } else {
+                    null
+                }
                 return if (kafkaConfig.enableAutoCommit) {
-                    KafkaAutoCommitEventConsumer(kafkaConfig, log)
+                    KafkaAutoCommitEventConsumer(kafkaConfig, log, dlqService)
                 } else {
-                    KafkaManualCommitEventConsumer(kafkaConfig, log)
+                    KafkaManualCommitEventConsumer(kafkaConfig, log, dlqService)
                 }
             }
         }
@@ -153,7 +164,8 @@ data class KafkaTopicConfig(val commit: Boolean, val topicPartitionsMap: Map<Top
 }
 
 open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfiguration,
-                                        private val log: Log): StreamsEventConsumer(log) {
+                                        private val log: Log,
+                                        private val dlqService: KafkaDLQService?): StreamsEventConsumer(log, dlqService) {
 
     private var isSeekSet = false
 
@@ -176,6 +188,7 @@ open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfigurati
 
     override fun stop() {
         consumer.close()
+        dlqService?.close()
     }
 
     private fun readSimple(action: (String, List<Any>) -> Unit) {
@@ -190,7 +203,9 @@ open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfigurati
         try {
             action(topic, convert(topicRecords))
         } catch (e: Exception) {
-            // TODO send to the DLQ
+            topicRecords
+                    .map { DLQData.from(it, e, this::class.java) }
+                    .forEach{ sentToDLQ(it) }
         }
     }
 
@@ -199,15 +214,20 @@ open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfigurati
                 try {
                     "ok" to JSONUtils.readValue<Any>(it.value())
                 } catch (e: Exception) {
-                    "error" to it
+                    "error" to DLQData.from(it, e, this::class.java)
                 }
             }
             .groupBy({ it.first }, { it.second })
             .let {
-                // TODO send content of the "error" key to the DLQ
+                it.getOrDefault("error", emptyList<DLQData>())
+                        .forEach{ sentToDLQ(it as DLQData) }
                 it.getOrDefault("ok", emptyList())
             }
 
+    private fun sentToDLQ(dlqData: DLQData) {
+        dlqService?.send(config.streamsSinkConfiguration.dlqTopic, dlqData)
+    }
+
     private fun readFromPartition(config: KafkaTopicConfig, action: (String, List<Any>) -> Unit) {
         setSeek(config.topicPartitionsMap)
         val records = consumer.poll(0)
@@ -248,7 +268,10 @@ open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfigurati
 }
 
 class KafkaManualCommitEventConsumer(private val config: KafkaSinkConfiguration,
-                                     private val log: Log): KafkaAutoCommitEventConsumer(config, log) {
+                                     private val log: Log,
+                                     private val dlqService: KafkaDLQService?): KafkaAutoCommitEventConsumer(config, log, dlqService) {
+
+    private val topicPartitionOffsetMap = ConcurrentHashMap<TopicPartition, OffsetAndMetadata>()
 
     override fun start() {
         if (topics.isEmpty()) {
@@ -295,6 +318,7 @@ class KafkaManualCommitEventConsumer(private val config: KafkaSinkConfiguration,
 
     override fun read(action: (String, List<Any>) -> Unit) {
         val topicMap = readSimple(action)
+        topicPartitionOffsetMap += topicMap
         commitData(true, topicMap)
     }
 
@@ -305,6 +329,7 @@ class KafkaManualCommitEventConsumer(private val config: KafkaSinkConfiguration,
         } else {
             readFromPartition(kafkaTopicConfig, action)
         }
+        topicPartitionOffsetMap += topicMap
         commitData(kafkaTopicConfig.commit, topicMap)
     }
 }
diff --git a/consumer/src/test/kotlin/integrations/KafkaEventSinkIT.kt b/consumer/src/test/kotlin/integrations/KafkaEventSinkIT.kt
diff --git a/doc/asciidoc/consumer/index.adoc b/doc/asciidoc/consumer/index.adoc
