Kafka Connect Adaptor: handle null offsets from kafka connector (#14916)

Author: dlg99

pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStore.java

@@ -213,8 +213,17 @@ public Future<Void> set(Map<ByteBuffer, ByteBuffer> values, Callback<Void> callb
         values.forEach((key, value) -> {
             ByteBuf bb = Unpooled.wrappedBuffer(key);
             byte[] keyBytes = ByteBufUtil.getBytes(bb);
-            bb = Unpooled.wrappedBuffer(value);
-            byte[] valBytes = ByteBufUtil.getBytes(bb);
+            byte[] valBytes = null;
+            if (value != null) {
+                bb = Unpooled.wrappedBuffer(value);
+                valBytes = ByteBufUtil.getBytes(bb);
+            } else {
+                // It does not actually matter if it is earliest or latest.
+                // The connector that provides null offsets works with the
+                // system that cannot seek to the offset anyway.
+                // Just need to store something to keep the offset store happy.
+                valBytes = MessageId.earliest.toByteArray();
+            }
             producer.newMessage()
                 .key(new String(keyBytes, UTF_8))
                 .value(valBytes)

pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/PulsarOffsetBackingStoreTest.java

@@ -35,6 +35,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.connect.util.Callback;
+import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.ProducerConsumerBase;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.testng.annotations.AfterMethod;
@@ -88,6 +89,28 @@ public void testGetFromEmpty() throws Exception {
         ).get().isEmpty());
     }
 
+    @Test(timeOut = 60000)
+    public void testGetSetNullValue() throws Exception {
+        Map<ByteBuffer, ByteBuffer> kvs = new HashMap<>();
+        ByteBuffer keyToSet = ByteBuffer.wrap(("test-key").getBytes(UTF_8));
+        kvs.put(keyToSet, null);
+        offsetBackingStore.set(kvs, null).get();
+
+        final List<ByteBuffer> keys = new ArrayList<>();
+        keys.add(keyToSet);
+
+        Map<ByteBuffer, ByteBuffer> result =
+                offsetBackingStore.get(keys).get();
+        assertEquals(1, result.size());
+
+        result.forEach((key, value) -> {
+            byte[] keyData = ByteBufUtil.getBytes(Unpooled.wrappedBuffer(key));
+            assertEquals(new String(keyData, UTF_8), "test-key");
+            byte[] valData = ByteBufUtil.getBytes(Unpooled.wrappedBuffer(value));
+            assertEquals(valData, MessageId.earliest.toByteArray());
+        });
+    }
+
     @Test
     public void testGetSet() throws Exception {
         testGetSet(false);