fixes #241: Temporary disable DLQ from the connect plugin (#242)

Author: conker84

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

@@ -17,7 +17,7 @@ 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<*>)
@@ -28,27 +28,27 @@ class Neo4jSinkTask : SinkTask() {
 
         val kafkaConfig = Properties()
         kafkaConfig.putAll(map)
-        this.errorService = KafkaErrorService(kafkaConfig, ErrorService.ErrorConfig.from(kafkaConfig), log::error)
+//        this.errorService = KafkaErrorService(kafkaConfig, ErrorService.ErrorConfig.from(kafkaConfig), log::error)
         this.neo4jService = Neo4jService(this.config)
     }
 
     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() {

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

@@ -9,6 +9,7 @@ import org.apache.kafka.connect.sink.SinkTask
 import org.apache.kafka.connect.sink.SinkTaskContext
 import org.junit.After
 import org.junit.Before
+import org.junit.Ignore
 import org.junit.Test
 import org.mockito.Mockito.mock
 import org.neo4j.graphdb.Label
@@ -388,6 +389,7 @@ class Neo4jSinkTaskTest {
     }
 
     @Test
+    @Ignore("temporary disabled the DLQ")
     fun `should report but not fail parsing data`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()
@@ -410,6 +412,7 @@ class Neo4jSinkTaskTest {
     }
 
     @Test
+    @Ignore("temporary disabled the DLQ")
     fun `should report but not fail invalid schema`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()
@@ -432,6 +435,7 @@ class Neo4jSinkTaskTest {
     }
 
     @Test
+    @Ignore("temporary disabled the DLQ")
     fun `should fail running invalid cypher`() {
         val topic = "neotopic"
         val props = mutableMapOf<String, String>()