simplify offset tracker

questdb · Oct 31, 2023 · dcf3275 · dcf3275
1 parent 1dca9f4
commit dcf3275
Show file tree

Hide file tree

Showing 7 changed files with 39 additions and 103 deletions.
diff --git a/connector/src/main/java/io/questdb/kafka/EmptyOffsetTracker.java b/connector/src/main/java/io/questdb/kafka/EmptyOffsetTracker.java
@@ -0,0 +1,35 @@
+package io.questdb.kafka;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.connect.sink.SinkTaskContext;
+
+import java.util.Collection;
+import java.util.Map;
+
+public final class EmptyOffsetTracker implements OffsetTracker {
+    @Override
+    public void onPartitionsOpened(Collection<TopicPartition> partitions) {
+
+    }
+
+    @Override
+    public void onPartitionsClosed(Collection<TopicPartition> partitions) {
+
+    }
+
+    @Override
+    public void onObservedOffset(int partition, String topic, long offset) {
+
+    }
+
+    @Override
+    public void configureSafeOffsets(SinkTaskContext sinkTaskContext, long rewindOffset) {
+        assert rewindOffset == 0;
+    }
+
+    @Override
+    public void transformPreCommit(Map<TopicPartition, OffsetAndMetadata> currentOffsets, long rewindOffset) {
+        assert rewindOffset == 0;
+    }
+}
diff --git a/...fka/MultiTopicPartitionOffsetTracker.java → .../io/questdb/kafka/MultiOffsetTracker.java b/...fka/MultiTopicPartitionOffsetTracker.java → .../io/questdb/kafka/MultiOffsetTracker.java
@@ -10,7 +10,7 @@
 import java.util.Collection;
 import java.util.Map;
 
-public final class MultiTopicPartitionOffsetTracker implements TopicPartitionOffsetTracker {
+public final class MultiOffsetTracker implements OffsetTracker {
     private static final int EMPTY = -1;
     private static final int CLOSED = -2;
 

diff --git a/...db/kafka/TopicPartitionOffsetTracker.java → .../java/io/questdb/kafka/OffsetTracker.java b/...db/kafka/TopicPartitionOffsetTracker.java → .../java/io/questdb/kafka/OffsetTracker.java
@@ -7,7 +7,7 @@
 import java.util.Collection;
 import java.util.Map;
 
-public interface TopicPartitionOffsetTracker {
+public interface OffsetTracker {
     void onPartitionsOpened(Collection<TopicPartition> partitions);
 
     void onPartitionsClosed(Collection<TopicPartition> partitions);

diff --git a/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java b/connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java
@@ -38,7 +38,7 @@ public final class QuestDBSinkTask extends SinkTask {
     private long batchesSinceLastError = 0;
     private DateFormat dataFormat;
     private boolean kafkaTimestampsEnabled;
-    private final TopicPartitionOffsetTracker tracker = new MultiTopicPartitionOffsetTracker();
+    private final OffsetTracker tracker = new MultiOffsetTracker();
 //private final TopicPartitionOffsetTracker tracker = new SingleTopicPartitionOffsetTracker();
 
     @Override

diff --git a/connector/src/main/java/io/questdb/kafka/SingleTopicPartitionOffsetTracker.java b/connector/src/main/java/io/questdb/kafka/SingleTopicPartitionOffsetTracker.java
diff --git a/connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java b/connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java
@@ -24,7 +24,6 @@
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
-import org.testcontainers.containers.wait.strategy.ShellStrategy;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.math.BigDecimal;
@@ -66,34 +65,9 @@ public static void createContainer() {
     @AfterAll
     public static void stopContainer() {
         questDBContainer.stop();
-//        deleteFromContainer("questdb");
         Files.rmdir(io.questdb.std.str.Path.getThreadLocal(dbRoot.toAbsolutePath().toString()));
     }
 
-    private static void deleteFromContainer(String directory) {
-        GenericContainer<?> cleanup = new GenericContainer<>("alpine:3.18.4")
-                .withFileSystemBind(dbRoot.toAbsolutePath().toString(), "/var/lib/delete")
-                .withCommand("ls -l /var/lib/delete/")
-                .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("cleanup1")));
-        cleanup.start();
-        cleanup.stop();
-
-        cleanup = new GenericContainer<>("alpine:3.18.4")
-                .withFileSystemBind(dbRoot.toAbsolutePath().toString(), "/var/lib/delete")
-                .withCommand("rm -rf /var/lib/delete/" + directory)
-                .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("cleanup2")))
-                .waitingFor(new ShellStrategy().withCommand("rm -rf /var/lib/delete/" + directory));
-        cleanup.start();
-        cleanup.stop();
-
-        cleanup = new GenericContainer<>("alpine:3.18.4")
-                .withFileSystemBind(dbRoot.toAbsolutePath().toString(), "/var/lib/delete")
-                .withCommand("ls -l /var/lib/delete/")
-                .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("cleanup3")));
-        cleanup.start();
-        cleanup.stop();
-    }
-
     private static String questDBDirectory() {
         return dbRoot.resolve("questdb").toAbsolutePath().toString();
     }

diff --git a/integration-tests/exactlyonce/src/test/java/io/questdb/kafka/ExactlyOnceIT.java b/integration-tests/exactlyonce/src/test/java/io/questdb/kafka/ExactlyOnceIT.java
@@ -246,8 +246,7 @@ private static String newPayload() {
         UUID uuid = UUID.randomUUID();
         int val = ThreadLocalRandom.current().nextInt(100);
 
-        String jsonVal = "{\"ts\":" + nanoTs + ",\"id\":\"" + uuid + "\",\"val\":" + val + "}";
-        return jsonVal;
+        return "{\"ts\":" + nanoTs + ",\"id\":\"" + uuid + "\",\"val\":" + val + "}";
     }
 
     private static void startKillingRandomContainers(CyclicBarrier barrier) {