feat: DQL for client-side validations failures

Author: jerrinot

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

@@ -97,6 +97,24 @@ public Sender boolColumn(CharSequence name, boolean value) {
         return this;
     }
 
+    @Override
+    public void cancelRow() {
+        symbolColumnNames.clear();
+        symbolColumnValues.clear();
+        stringNames.clear();
+        stringValues.clear();
+        longNames.clear();
+        longValues.clear();
+        doubleNames.clear();
+        doubleValues.clear();
+        boolNames.clear();
+        boolValues.clear();
+        timestampNames.clear();
+        timestampValues.clear();
+
+        sender.cancelRow();
+    }
+
     @Override
     public Sender timestampColumn(CharSequence name, long value, ChronoUnit unit) {
         if (symbolColumns.contains(name)) {

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

@@ -0,0 +1,14 @@
+package io.questdb.kafka;
+
+import io.questdb.std.NumericException;
+import org.apache.kafka.connect.errors.ConnectException;
+
+public final class InvalidDataException extends ConnectException {
+    public InvalidDataException(String message) {
+        super(message);
+    }
+
+    public InvalidDataException(String message, NumericException e) {
+        super(message, e);
+    }
+}

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

@@ -172,7 +172,22 @@ public void put(Collection<SinkRecord> collection) {
                 if (httpTransport) {
                     inflightSinkRecords.add(record);
                 }
-                handleSingleRecord(record);
+                try {
+                    handleSingleRecord(record);
+                } catch (InvalidDataException ex) {
+                    // data format error generated on client-side
+
+                    if (httpTransport && reporter != null) {
+                        // we have DLQ set, let's report this single object
+
+                        // remove the last item from in-flight records
+                        inflightSinkRecords.setPos(inflightSinkRecords.size() - 1);
+                        context.errantRecordReporter().report(record, ex);
+                    } else {
+                        // ok, no DQL, let's error the connector
+                        throw ex;
+                    }
+                }
             }
 
             if (httpTransport) {
@@ -257,7 +272,7 @@ private void onTcpSenderException(Exception e) {
     private void onHttpSenderException(Exception e) {
         closeSenderSilently();
         if (
-                (reporter != null && e.getMessage() != null) // hack to detect data parsing errors
+                (reporter != null && e.getMessage() != null) // hack to detect data parsing errors originating at server-side
                 && (e.getMessage().contains("error in line") || e.getMessage().contains("failed to parse line protocol"))
         ) {
             // ok, we have a parsing error, let's try to send records one by one to find the problematic record
@@ -300,16 +315,27 @@ private void handleSingleRecord(SinkRecord record) {
         assert timestampColumnValue == Long.MIN_VALUE;
 
         CharSequence tableName = recordToTable.apply(record);
+        if (tableName == null || tableName.equals("")) {
+            throw new InvalidDataException("Table name cannot be empty");
+        }
         sender.table(tableName);
 
-        if (config.isIncludeKey()) {
-            handleObject(config.getKeyPrefix(), record.keySchema(), record.key(), PRIMITIVE_KEY_FALLBACK_NAME);
+        try {
+            if (config.isIncludeKey()) {
+                handleObject(config.getKeyPrefix(), record.keySchema(), record.key(), PRIMITIVE_KEY_FALLBACK_NAME);
+            }
+            handleObject(config.getValuePrefix(), record.valueSchema(), record.value(), PRIMITIVE_VALUE_FALLBACK_NAME);
+        } catch (InvalidDataException ex) {
+            if (httpTransport) {
+                sender.cancelRow();
+            }
+            throw ex;
         }
-        handleObject(config.getValuePrefix(), record.valueSchema(), record.value(), PRIMITIVE_VALUE_FALLBACK_NAME);
 
         if (kafkaTimestampsEnabled) {
             timestampColumnValue = TimeUnit.MILLISECONDS.toNanos(record.timestamp());
         }
+
         if (timestampColumnValue == Long.MIN_VALUE) {
             sender.atNow();
         } else {
@@ -338,7 +364,7 @@ private void handleMap(String name, Map<?, ?> value, String fallbackName) {
         for (Map.Entry<?, ?> entry : value.entrySet()) {
             Object mapKey = entry.getKey();
             if (!(mapKey instanceof String)) {
-                throw new ConnectException("Map keys must be strings");
+                throw new InvalidDataException("Map keys must be strings");
             }
             String mapKeyName = (String) mapKey;
             String entryName = name.isEmpty() ? mapKeyName : name + STRUCT_FIELD_SEPARATOR + mapKeyName;
@@ -365,7 +391,7 @@ private void handleObject(String name, Schema schema, Object value, String fallb
         if (isDesignatedColumnName(name, fallbackName)) {
             assert timestampColumnValue == Long.MIN_VALUE;
             if (value == null) {
-                throw new ConnectException("Timestamp column value cannot be null");
+                throw new InvalidDataException("Timestamp column value cannot be null");
             }
             timestampColumnValue = resolveDesignatedTimestampColumnValue(value, schema);
             return;
@@ -393,7 +419,7 @@ private long resolveDesignatedTimestampColumnValue(Object value, Schema schema)
             return parseToMicros((String) value) * 1000;
         }
         if (!(value instanceof Long)) {
-            throw new ConnectException("Unsupported timestamp column type: " + value.getClass());
+            throw new InvalidDataException("Unsupported timestamp column type: " + value.getClass());
         }
         long longValue = (Long) value;
         TimeUnit inputUnit;
@@ -453,7 +479,7 @@ private long parseToMicros(String timestamp) {
         try {
             return dataFormat.parse(timestamp, DateFormatUtils.EN_LOCALE);
         } catch (NumericException e) {
-            throw new ConnectException("Cannot parse timestamp: " + timestamp + " with the configured format '" + config.getTimestampFormat() +"' use '"
+            throw new InvalidDataException("Cannot parse timestamp: " + timestamp + " with the configured format '" + config.getTimestampFormat() +"' use '"
                     + QuestDBSinkConnectorConfig.TIMESTAMP_FORMAT + "' to configure the right timestamp format. " +
                     "See https://questdb.io/docs/reference/function/date-time/#date-and-timestamp-format for timestamp parser documentation. ", e);
         }
@@ -513,7 +539,7 @@ private void onUnsupportedType(String name, Object type) {
         if (config.isSkipUnsupportedTypes()) {
             log.debug("Skipping unsupported type: {}, name: {}", type, name);
         } else {
-            throw new ConnectException("Unsupported type: " + type + ", name: " + name);
+            throw new InvalidDataException("Unsupported type: " + type + ", name: " + name);
         }
     }
 

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:8.1.1";
+    private static final String OFFICIAL_QUESTDB_DOCKER = "questdb/questdb:8.2.0";
     private static final boolean DUMP_QUESTDB_CONTAINER_LOGS = true;
 
     private EmbeddedConnectCluster connect;

pom.xml

@@ -84,7 +84,7 @@
         <dependency>
             <groupId>org.questdb</groupId>
             <artifactId>questdb</artifactId>
-            <version>7.4.0</version>
+            <version>8.2.0</version>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>