fix: handle empty input gracefully

Author: jerrinot

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

@@ -90,6 +90,10 @@ private Sender createSender() {
 
     @Override
     public void put(Collection<SinkRecord> collection) {
+        if (collection.isEmpty()) {
+            log.debug("Received empty collection, ignoring");
+            return;
+        }
         if (log.isDebugEnabled()) {
             SinkRecord record = collection.iterator().next();
             log.debug("Received {} records. First record kafka coordinates:({}-{}-{}). ",

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

@@ -259,6 +259,37 @@ public void testRetrying_recoversFromInfrastructureIssues() throws Exception {
                 "select firstname,lastname,age from " + topicName + " where age = 49");
     }
 
+    @Test
+    public void testEmptyCollection_wontFailTheConnector() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        // filter out all message
+        props.put("transforms", "drop");
+        props.put("transforms.drop.type", "org.apache.kafka.connect.transforms.Filter");
+
+
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+        Schema schema = SchemaBuilder.struct().name("com.example.Person")
+                .field("firstname", Schema.STRING_SCHEMA)
+                .field("lastname", Schema.STRING_SCHEMA)
+                .field("age", Schema.INT8_SCHEMA)
+                .build();
+
+        Struct struct = new Struct(schema)
+                .put("firstname", "John")
+                .put("lastname", "Doe")
+                .put("age", (byte) 42);
+
+        connect.kafka().produce(topicName, "key", new String(converter.fromConnectData(topicName, schema, struct)));
+
+        int durationMs = 10_000;
+        long deadline = System.currentTimeMillis() + durationMs;
+        while (System.currentTimeMillis() < deadline) {
+            assertConnectorTaskState(CONNECTOR_NAME, AbstractStatus.State.RUNNING);
+        }
+    }
+
     @Test
     public void testSymbol_withAllOtherILPTypes() {
         connect.kafka().createTopic(topicName, 1);

connector/src/test/resources/log4j.properties

@@ -26,3 +26,4 @@ log4j.logger.org.apache.zookeeper=WARN
 log4j.logger.kafka=WARN
 log4j.logger.org.reflections=ERROR
 log4j.logger.state.change.logger=WARN
+log4j.logger.io.questdb.kafka=DEBUG