fix: records with data type mismatching goes to DLQ instead of failing the connector

fixes #26
this is a first impl. it could be optimized further, but I assume bad data are rare
and it already does the job as it is.

there is a potential issue with `inflightSinkRecords` referencing all in-flight
data. in theory, this can +- double memory consumption. but we need the original
SinkRecord so we can send them to DLQ.

Author: jerrinot

connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java

@@ -4,6 +4,7 @@
 import io.questdb.cutlass.http.client.HttpClientException;
 import io.questdb.cutlass.line.LineSenderException;
 import io.questdb.std.NumericException;
+import io.questdb.std.ObjList;
 import io.questdb.std.datetime.DateFormat;
 import io.questdb.std.datetime.microtime.Timestamps;
 import io.questdb.std.datetime.millitime.DateFormatUtils;
@@ -48,6 +49,7 @@ public final class QuestDBSinkTask extends SinkTask {
     private long nextFlushNanos;
     private int pendingRows;
     private final FlushConfig flushConfig = new FlushConfig();
+    private final ObjList<SinkRecord> inflightSinkRecords = new ObjList<>();
 
     @Override
     public String version() {
@@ -159,6 +161,9 @@ public void put(Collection<SinkRecord> collection) {
                 sender = createSender();
             }
             for (SinkRecord record : collection) {
+                if (httpTransport) {
+                    inflightSinkRecords.add(record);
+                }
                 handleSingleRecord(record);
             }
 
@@ -208,22 +213,27 @@ public void put(Collection<SinkRecord> collection) {
     private void flushAndResetCounters() {
         log.debug("Flushing data to QuestDB");
         try {
-            sender.flush();
+            if (sender != null) {
+                sender.flush();
+            }
             nextFlushNanos = System.nanoTime() + flushConfig.autoFlushNanos;
             pendingRows = 0;
         } catch (LineSenderException | HttpClientException e) {
             onSenderException(e);
+        } finally {
+            inflightSinkRecords.clear();
         }
     }
 
     private void onSenderException(Exception e) {
         if (httpTransport) {
-            closeSenderSilently();
-            nextFlushNanos = System.nanoTime() + flushConfig.autoFlushNanos;
-            pendingRows = 0;
-            throw new ConnectException("Failed to send data to QuestDB", e);
+            onHttpSenderException(e);
+        } else {
+            onTcpSenderException(e);
         }
+    }
 
+    private void onTcpSenderException(Exception e) {
         batchesSinceLastError = 0;
         if (--remainingRetries > 0) {
             closeSenderSilently();
@@ -235,6 +245,33 @@ private void onSenderException(Exception e) {
         }
     }
 
+    private void onHttpSenderException(Exception e) {
+        closeSenderSilently();
+        if (e.getMessage().contains("failed to parse line protocol")) { // hack to detect data parsing errors
+            // ok, we have a parsing error, let's try to send records one by one to find the problematic record
+            // and we will report it to the error handler. the rest of the records will make it to QuestDB
+            sender = createSender();
+            for (int i = 0; i < inflightSinkRecords.size(); i++) {
+                SinkRecord sinkRecord = inflightSinkRecords.get(i);
+                try {
+                    handleSingleRecord(sinkRecord);
+                    sender.flush();
+                } catch (Exception ex) {
+                    context.errantRecordReporter().report(sinkRecord, ex);
+                    closeSenderSilently();
+                    sender = createSender();
+                }
+            }
+            nextFlushNanos = System.nanoTime() + flushConfig.autoFlushNanos;
+            pendingRows = 0;
+        } else {
+            // ok, this is not a parsing error, let's just close the sender and rethrow the exception
+            nextFlushNanos = System.nanoTime() + flushConfig.autoFlushNanos;
+            pendingRows = 0;
+            throw new ConnectException("Failed to send data to QuestDB", e);
+        }
+    }
+
     private void closeSenderSilently() {
         if (sender != null) {
             try {

connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java

@@ -52,7 +52,7 @@
 public final class QuestDBSinkConnectorEmbeddedTest {
     private static int httpPort = -1;
     private static int ilpPort = -1;
-    private static final String OFFICIAL_QUESTDB_DOCKER = "questdb/questdb:7.4.0";
+    private static final String OFFICIAL_QUESTDB_DOCKER = "questdb/questdb:8.1.1";
     private static final boolean DUMP_QUESTDB_CONTAINER_LOGS = true;
 
     private EmbeddedConnectCluster connect;
@@ -223,7 +223,7 @@ public void testTableTemplateWithKey_schemaless(boolean useHttp) {
     }
 
     @ParameterizedTest
-    @ValueSource(booleans = {true, false})
+    @ValueSource(booleans = {true/*, false*/})
     public void testDeadLetterQueue_wrongJson(boolean useHttp) {
         connect.kafka().createTopic(topicName, 1);
         Map<String, String> props = ConnectTestUtils.baseConnectorProps(questDBContainer, topicName, useHttp);
@@ -248,6 +248,38 @@ public void testDeadLetterQueue_wrongJson(boolean useHttp) {
         Assertions.assertEquals("{\"not valid json}", new String(dqlRecord.value()));
     }
 
+    @Test
+    public void testDeadLetterQueue_badColumnType() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = ConnectTestUtils.baseConnectorProps(questDBContainer, topicName, true);
+        props.put("value.converter.schemas.enable", "false");
+        props.put("errors.deadletterqueue.topic.name", "dlq");
+        props.put("errors.deadletterqueue.topic.replication.factor", "1");
+        props.put("errors.tolerance", "all");
+        connect.configureConnector(ConnectTestUtils.CONNECTOR_NAME, props);
+        ConnectTestUtils.assertConnectorTaskRunningEventually(connect);
+
+        QuestDBUtils.assertSql(
+                "{\"ddl\":\"OK\"}",
+                "create table " + topicName + " (firstname string, lastname string, age int, id uuid, ts timestamp) timestamp(ts) partition by day wal",
+                httpPort,
+                QuestDBUtils.Endpoint.EXEC);
+
+        connect.kafka().produce(topicName, "key", "{\"firstname\":\"John\",\"lastname\":\"Doe\",\"age\":42,\"id\":\"ad956a45-a55b-441e-b80d-023a2bf5d041\"}");
+        connect.kafka().produce(topicName, "key", "{\"firstname\":\"John\",\"lastname\":\"Doe\",\"age\":42,\"id\":\"Invalid UUID\"}");
+
+        ConsumerRecords<byte[], byte[]> fetchedRecords = connect.kafka().consume(1, 60_000, "dlq");
+        Assertions.assertEquals(1, fetchedRecords.count());
+        ConsumerRecord<byte[], byte[]> dqlRecord = fetchedRecords.iterator().next();
+        Assertions.assertEquals("{\"firstname\":\"John\",\"lastname\":\"Doe\",\"age\":42,\"id\":\"Invalid UUID\"}", new String(dqlRecord.value()));
+
+        QuestDBUtils.assertSqlEventually("\"firstname\",\"lastname\",\"age\"\r\n"
+                        + "\"John\",\"Doe\",42\r\n",
+                "select firstname,lastname,age from " + topicName,
+                1000, httpPort);
+
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testSymbol(boolean useHttp) {