Fixes #356: added a database field to the message in the DLQ (#448)

vga91 · web-flow · commit 0b177be55ba4 · 2021-03-26T09:34:13.000-04:00
* fixes #356 * changed empty handling
diff --git a/common/src/main/kotlin/streams/service/errors/ErrorService.kt b/common/src/main/kotlin/streams/service/errors/ErrorService.kt
@@ -15,23 +15,25 @@ data class ErrorData(val originalTopic: String,
                      val partition: String,
                      val offset: String,
                      val executingClass: Class<*>?,
+                     val databaseName: String?,
                      val exception: Exception?) {
 
     constructor(originalTopic: String, timestamp: Long?, key: Any?, value: Any?,
-                partition: Int, offset: Long, executingClass: Class<*>?, exception: Exception?) :
-            this(originalTopic, timestamp ?: RecordBatch.NO_TIMESTAMP, toByteArray(key), toByteArray(value), partition.toString(),offset.toString(), executingClass, exception)
+                partition: Int, offset: Long, executingClass: Class<*>?, databaseName: String?, exception: Exception?) :
+            this(originalTopic, timestamp ?: RecordBatch.NO_TIMESTAMP, toByteArray(key), toByteArray(value), partition.toString(),offset.toString(), executingClass, databaseName, exception)
 
     companion object {
 
-        fun from(consumerRecord: ConsumerRecord<out Any, out Any>, exception: Exception?, executingClass: Class<*>?): ErrorData {
+        fun from(consumerRecord: ConsumerRecord<out Any, out Any>, exception: Exception?, executingClass: Class<*>?, databaseName: String?): ErrorData {
             return ErrorData(offset = consumerRecord.offset().toString(),
                     originalTopic = consumerRecord.topic(),
                     partition = consumerRecord.partition().toString(),
                     timestamp = consumerRecord.timestamp(),
                     exception = exception,
                     executingClass = executingClass,
                     key = toByteArray(consumerRecord.key()),
-                    value = toByteArray(consumerRecord.value()))
+                    value = toByteArray(consumerRecord.value()),
+                    databaseName = databaseName)
         }
 
         fun toByteArray(v:Any?) = try {
diff --git a/common/src/main/kotlin/streams/service/errors/KafkaErrorService.kt b/common/src/main/kotlin/streams/service/errors/KafkaErrorService.kt
@@ -72,6 +72,10 @@ class KafkaErrorService(private val producer: Producer<ByteArray, ByteArray>?, p
         prefix("partition") to errorData.partition.toByteArray(),
         prefix("offset") to errorData.offset.toByteArray())
 
+        if (!errorData.databaseName.isNullOrBlank()) {
+            headers[prefix("databaseName")] = errorData.databaseName.toByteArray()
+        }
+
         if (errorData.executingClass != null) {
             headers[prefix("class.name")] = errorData.executingClass.name.toByteArray()
         }
diff --git a/common/src/test/kotlin/streams/service/sink/errors/KafkaErrorServiceTest.kt b/common/src/test/kotlin/streams/service/sink/errors/KafkaErrorServiceTest.kt
@@ -59,6 +59,7 @@ class KafkaErrorServiceTest {
         val exception = RuntimeException("Test")
         val key = "KEY"
         val value = "VALUE"
+        val databaseName = "myDb"
         return ErrorData(
                 offset = offset,
                 originalTopic = originalTopic,
@@ -67,7 +68,8 @@ class KafkaErrorServiceTest {
                 exception = exception,
                 executingClass = KafkaErrorServiceTest::class.java,
                 key = key.toByteArray(),
-                value = value.toByteArray()
+                value = value.toByteArray(),
+                databaseName = databaseName
         )
     }
 
diff --git a/consumer/src/main/kotlin/streams/kafka/KafkaAutoCommitEventConsumer.kt b/consumer/src/main/kotlin/streams/kafka/KafkaAutoCommitEventConsumer.kt
@@ -49,7 +49,8 @@ abstract class KafkaEventConsumer(config: KafkaSinkConfiguration,
 
 open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfiguration,
                                         private val log: Log,
-                                        val topics: Set<String>): KafkaEventConsumer(config, log, topics) {
+                                        val topics: Set<String>,
+                                        private val dbName: String): KafkaEventConsumer(config, log, topics) {
 
     private val errorService: ErrorService = KafkaErrorService(config.asProperties(),
         ErrorService.ErrorConfig.from(config.streamsSinkConfiguration.errorConfig),
@@ -93,7 +94,7 @@ open class KafkaAutoCommitEventConsumer(private val config: KafkaSinkConfigurati
         try {
             action(topic, topicRecords.map { it.toStreamsSinkEntity() })
         } catch (e: Exception) {
-            errorService.report(topicRecords.map { ErrorData.from(it, e, this::class.java) })
+            errorService.report(topicRecords.map { ErrorData.from(it, e, this::class.java, dbName) })
         }
     }
 
diff --git a/consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt b/consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt
@@ -15,7 +15,12 @@ import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.errors.WakeupException
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.logging.Log
-import streams.*
+import streams.StreamsEventConsumer
+import streams.StreamsEventConsumerFactory
+import streams.StreamsEventSink
+import streams.StreamsEventSinkQueryExecution
+import streams.StreamsSinkConfiguration
+import streams.StreamsTopicService
 import streams.config.StreamsConfig
 import streams.events.StreamsPluginStatus
 import streams.extensions.isDefaultDb
@@ -55,12 +60,13 @@ class KafkaEventSink(private val config: Map<String, String>,
     override fun getEventConsumerFactory(): StreamsEventConsumerFactory {
         return object: StreamsEventConsumerFactory() {
             override fun createStreamsEventConsumer(config: Map<String, String>, log: Log, topics: Set<Any>): StreamsEventConsumer {
-                val kafkaConfig = KafkaSinkConfiguration.from(config, db.databaseName(), db.isDefaultDb())
+                val dbName = db.databaseName()
+                val kafkaConfig = KafkaSinkConfiguration.from(config, dbName, db.isDefaultDb())
                 val topics1 = topics as Set<String>
                 return if (kafkaConfig.enableAutoCommit) {
-                    KafkaAutoCommitEventConsumer(kafkaConfig, log, topics1)
+                    KafkaAutoCommitEventConsumer(kafkaConfig, log, topics1, dbName)
                 } else {
-                    KafkaManualCommitEventConsumer(kafkaConfig, log, topics1)
+                    KafkaManualCommitEventConsumer(kafkaConfig, log, topics1, dbName)
                 }
             }
         }
diff --git a/consumer/src/main/kotlin/streams/kafka/KafkaManualCommitEventConsumer.kt b/consumer/src/main/kotlin/streams/kafka/KafkaManualCommitEventConsumer.kt
@@ -13,7 +13,8 @@ import java.time.Duration
 
 class KafkaManualCommitEventConsumer(config: KafkaSinkConfiguration,
                                      private val log: Log,
-                                     topics: Set<String>): KafkaAutoCommitEventConsumer(config, log, topics) {
+                                     topics: Set<String>,
+                                     dbName: String): KafkaAutoCommitEventConsumer(config, log, topics, dbName) {
 
     private val asyncCommit = config.streamsAsyncCommit
 
diff --git a/consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkDLQTSE.kt b/consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkDLQTSE.kt
@@ -28,7 +28,7 @@ class KafkaEventSinkDLQTSE : KafkaEventSinkBaseTSE() {
         db.start()
         val data = mapOf("id" to null, "name" to "Andrea", "surname" to "Santurbano")
 
-        var producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(data))
+        val producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(data))
         kafkaProducer.send(producerRecord).get()
         val dlqConsumer = KafkaTestUtils.createConsumer<ByteArray, ByteArray>(
                 bootstrapServers = KafkaEventSinkSuiteIT.kafka.bootstrapServers,
@@ -50,8 +50,9 @@ class KafkaEventSinkDLQTSE : KafkaEventSinkBaseTSE() {
                 val value = if (record != null) JSONUtils.readValue<Any>(record.value()!!) else emptyMap<String, Any>()
                 db.execute(query) {
                     val result = it.columnAs<Long>("count")
-                    !records.isEmpty && headers.size == 7 && value == data && result.hasNext() && result.next() == 0L && !result.hasNext()
+                    !records.isEmpty && headers.size == 8 && value == data && result.hasNext() && result.next() == 0L && !result.hasNext()
                             && headers["__streams.errors.exception.class.name"] == "org.neo4j.graphdb.QueryExecutionException"
+                            && headers["__streams.errors.databaseName"] == "neo4j"
                 }
             }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
         }
@@ -70,7 +71,7 @@ class KafkaEventSinkDLQTSE : KafkaEventSinkBaseTSE() {
 
         val data = """{id: 1, "name": "Andrea", "surname": "Santurbano"}"""
 
-        var producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(),
+        val producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(),
                 data.toByteArray())
         kafkaProducer.send(producerRecord).get()
         val dlqConsumer = KafkaTestUtils.createConsumer<ByteArray, ByteArray>(
@@ -91,8 +92,9 @@ class KafkaEventSinkDLQTSE : KafkaEventSinkBaseTSE() {
                     val record = if (records.isEmpty) null else records.records(dlqTopic).iterator().next()
                     val headers = record?.headers()?.map { it.key() to String(it.value()) }?.toMap().orEmpty()
                     val value = if (record != null) String(record.value()) else emptyMap<String, Any>()
-                    !records.isEmpty && headers.size == 7 && data == value && count.hasNext() && count.next() == 0L && !count.hasNext()
+                    !records.isEmpty && headers.size == 8 && data == value && count.hasNext() && count.next() == 0L && !count.hasNext()
                             && headers["__streams.errors.exception.class.name"] == "com.fasterxml.jackson.core.JsonParseException"
+                            && headers["__streams.errors.databaseName"] == "neo4j"
                 }
             }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
         }
diff --git a/consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkEnterpriseTSE.kt b/consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkEnterpriseTSE.kt
@@ -6,6 +6,7 @@ import kotlinx.coroutines.runBlocking
 import org.apache.avro.generic.GenericRecord
 import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.hamcrest.Matchers
 import org.junit.After
 import org.junit.AfterClass
@@ -17,8 +18,8 @@ import org.neo4j.driver.SessionConfig
 import org.neo4j.function.ThrowingSupplier
 import streams.Assert
 import streams.KafkaTestUtils
-import streams.MavenUtils
 import streams.Neo4jContainerExtension
+import streams.service.errors.ErrorService
 import streams.utils.JSONUtils
 import streams.utils.StreamsUtils
 import java.util.UUID
@@ -29,8 +30,10 @@ class KafkaEventSinkEnterpriseTSE {
     companion object {
 
         private var startedFromSuite = true
-        val DB_NAME_NAMES = arrayOf("foo", "bar")
-        val ALL_DBS = arrayOf("foo", "bar", "baz")
+        private val DB_NAME_NAMES = arrayOf("foo", "bar", "dlq")
+        val ALL_DBS = arrayOf("foo", "bar", "baz", "dlq")
+        private const val DLQ_ERROR_TOPIC = "dlqTopic"
+        const val DLQ_CYPHER_TOPIC = "dlqCypherTopic"
 
         @JvmStatic
         val neo4j = Neo4jContainerExtension()//.withLogging()
@@ -50,6 +53,11 @@ class KafkaEventSinkEnterpriseTSE {
                 DB_NAME_NAMES.forEach { neo4j.withNeo4jConfig("streams.sink.enabled.to.$it", "true") } // we enable the sink plugin only for the instances
                 neo4j.withNeo4jConfig("streams.sink.topic.cypher.enterpriseCypherTopic.to.foo", "MERGE (c:Customer_foo {id: event.id, foo: 'foo'})")
                 neo4j.withNeo4jConfig("streams.sink.topic.cypher.enterpriseCypherTopic.to.bar", "MERGE (c:Customer_bar {id: event.id, bar: 'bar'})")
+                neo4j.withNeo4jConfig("streams.sink.topic.cypher.$DLQ_CYPHER_TOPIC.to.dlq", "MERGE (c:Customer_dlq {id: event.id, dlq: 'dlq'})")
+                neo4j.withNeo4jConfig("streams.sink." + ErrorService.ErrorConfig.DLQ_TOPIC, DLQ_ERROR_TOPIC)
+                neo4j.withNeo4jConfig("streams.sink." + ErrorService.ErrorConfig.DLQ_HEADERS, "true")
+                neo4j.withNeo4jConfig("streams.sink." + ErrorService.ErrorConfig.DLQ_HEADER_PREFIX, "__streams.errors.")
+                neo4j.withNeo4jConfig("streams.sink." + ErrorService.ErrorConfig.TOLERANCE, "all")
                 neo4j.withDatabases(*ALL_DBS)
                 neo4j.start()
                 Assume.assumeTrue("Neo4j must be running", neo4j.isRunning)
@@ -126,4 +134,39 @@ class KafkaEventSinkEnterpriseTSE {
             nodes.isEmpty()
         }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
     }
+
+    @Test
+    fun `should send data to the DLQ with current databaseName because of JsonParseException`() = runBlocking {
+
+        val data = mapOf("id" to null, "name" to "Andrea", "surname" to "Santurbano")
+
+        val producerRecord = ProducerRecord(DLQ_CYPHER_TOPIC, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(data))
+
+        // when
+        kafkaProducer.send(producerRecord).get()
+        delay(5000)
+
+        val dlqConsumer = KafkaTestUtils.createConsumer<ByteArray, ByteArray>(
+                bootstrapServers = KafkaEventSinkSuiteIT.kafka.bootstrapServers,
+                schemaRegistryUrl = KafkaEventSinkSuiteIT.schemaRegistry.getSchemaRegistryUrl(),
+                keyDeserializer = ByteArrayDeserializer::class.java.name,
+                valueDeserializer = ByteArrayDeserializer::class.java.name,
+                topics = arrayOf(DLQ_ERROR_TOPIC))
+
+        dlqConsumer.let {
+            Assert.assertEventually(ThrowingSupplier {
+                val dbName = "dlq"
+                val nodes = getData(dbName)
+                val count = nodes.size
+                val records = dlqConsumer.poll(5000)
+                val record = if (records.isEmpty) null else records.records(DLQ_ERROR_TOPIC).iterator().next()
+                val headers = record?.headers()?.map { it.key() to String(it.value()) }?.toMap().orEmpty()
+                val value = if (record != null) JSONUtils.readValue<Any>(record.value()!!) else emptyMap<String, Any>()
+                !records.isEmpty && headers.size == 8 && data == value && count == 0
+                        && headers["__streams.errors.exception.class.name"] == "org.neo4j.graphdb.QueryExecutionException"
+                        && headers["__streams.errors.databaseName"] == dbName
+            }, Matchers.equalTo(true), 30, TimeUnit.SECONDS)
+            it.close()
+        }
+    }
 }
diff --git a/doc/docs/modules/ROOT/pages/consumer.adoc b/doc/docs/modules/ROOT/pages/consumer.adoc
@@ -190,6 +190,9 @@ Every published record in the `Dead Letter Queue` contains the original record `
 
 | <prefix>exception.stacktrace"
 | The exception stack trace
+
+| <prefix>databaseName"
+| The database name
 |===
 
 [[neo4j_streams_supported_deserializers]]
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
@@ -45,7 +45,7 @@ class Neo4jSinkTask : SinkTask() {
             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)
+                ErrorData(it.topic(), it.timestamp(), it.key(), it.value(), it.kafkaPartition(), it.kafkaOffset(), this::class.java, this.config.database, e)
             })
         }
     }
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
@@ -17,6 +17,7 @@ import org.neo4j.harness.junit.rule.Neo4jRule
 import streams.events.*
 import streams.utils.JSONUtils
 import streams.service.errors.ErrorService
+import streams.service.errors.ProcessingError
 import streams.service.sink.strategy.CUDNode
 import streams.service.sink.strategy.CUDNodeRel
 import streams.service.sink.strategy.CUDOperations
@@ -28,6 +29,7 @@ import java.util.stream.StreamSupport
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
 import kotlin.test.assertTrue
+import kotlin.test.fail
 
 
 class Neo4jSinkTaskTest {
@@ -1227,4 +1229,52 @@ class Neo4jSinkTaskTest {
         }
     }
 
+    @Test
+    fun `should fail data insertion with ProcessingError`() {
+        // given
+        val topic = UUID.randomUUID().toString()
+
+        val props = mutableMapOf<String, String>()
+        props[Neo4jSinkConnectorConfig.SERVER_URI] = db.boltURI().toString()
+        props["${Neo4jSinkConnectorConfig.TOPIC_PATTERN_RELATIONSHIP_PREFIX}$topic"] = "(:User{!sourceId,sourceName,sourceSurname})-[:KNOWS]->(:User{!targetId,targetName,targetSurname})"
+        props[Neo4jSinkConnectorConfig.AUTHENTICATION_TYPE] = AuthenticationType.NONE.toString()
+        props[SinkTask.TOPICS_CONFIG] = topic
+        props[Neo4jSinkConnectorConfig.DATABASE] = "notExistent"
+
+        val data = mapOf("sourceId" to 1, "sourceName" to "Andrea", "sourceSurname" to "Santurbano",
+                "targetId" to 1, "targetName" to "Michael", "targetSurname" to "Hunger", "since" to 2014)
+
+        val task = Neo4jSinkTask()
+        task.initialize(mock(SinkTaskContext::class.java))
+        task.start(props)
+        val input = listOf(SinkRecord(topic, 1, null, null, null, data, 42))
+
+        try {
+            task.put(input)
+            fail("It should fail with ProcessingError")
+        } catch (e: ProcessingError) {
+            val errorData = e.errorDatas.first()
+            assertTrue(errorData.databaseName == "notExistent"
+                    && errorData.exception!!.javaClass.name == "org.neo4j.driver.exceptions.FatalDiscoveryException")
+        }
+
+        props[Neo4jSinkConnectorConfig.DATABASE] = "neo4j"
+        val taskNotValid = Neo4jSinkTask()
+        taskNotValid.initialize(mock(SinkTaskContext::class.java))
+        taskNotValid.start(props)
+
+        val dataNotValid = mapOf("sourceId" to null, "sourceName" to "Andrea", "sourceSurname" to "Santurbano",
+                "targetId" to 1, "targetName" to "Michael", "targetSurname" to "Hunger", "since" to 2014)
+        val inputNotValid = listOf(SinkRecord(topic, 1, null, null, null, dataNotValid, 43))
+
+        try {
+            taskNotValid.put(inputNotValid)
+            fail("It should fail with ProcessingError")
+        } catch (e: ProcessingError) {
+            val errorData = e.errorDatas.first()
+            assertTrue(errorData.databaseName == "neo4j"
+                    && errorData.exception!!.javaClass.name == "org.neo4j.driver.exceptions.ClientException")
+        }
+    }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,7 @@ class Neo4jSinkTask : SinkTask() {`
`45`	`45`	`neo4jService.writeData(data)`
`46`	`46`	`} catch(e:Exception) {`
`47`	`47`	`errorService.report(collection.map {`
`48`		`- ErrorData(it.topic(), it.timestamp(), it.key(), it.value(), it.kafkaPartition(), it.kafkaOffset(), this::class.java, e)`
	`48`	`+ ErrorData(it.topic(), it.timestamp(), it.key(), it.value(), it.kafkaPartition(), it.kafkaOffset(), this::class.java, this.config.database, e)`
`49`	`49`	`})`
`50`	`50`	`}`
`51`	`51`	`}`