fix(plugin): fix NPE using version 2.0 with KafkaFileObjectStateBackingStore (#149)

Resolves: #149

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/KafkaBasedLog.java

@@ -52,18 +52,18 @@ public class KafkaBasedLog<K, V> {
     private static final Logger LOG = LoggerFactory.getLogger(KafkaBasedLog.class);
     private static final long CREATE_TOPIC_TIMEOUT_MS = 30000;
 
-    private Time time;
+    private final Time time;
     private final String topic;
     private final Map<String, Object> producerConfigs;
     private final Map<String, Object> consumerConfigs;
     private final Callback<ConsumerRecord<K, V>> consumedCallback;
     private Consumer<K, V> consumer;
-    private Producer<K, V> producer;
+    private volatile Producer<K, V> producer;
 
     private Thread thread;
     private boolean stopRequested;
-    private Queue<Callback<Void>> readLogEndOffsetCallbacks;
-    private Runnable initializer;
+    private final Queue<Callback<Void>> readLogEndOffsetCallbacks;
+    private final Runnable initializer;
 
     private volatile States state;
 
@@ -230,14 +230,22 @@ public Future<Void> readToEnd() {
         return future;
     }
 
-    public void send(K key, V value) {
+    public void send(final K key, final V value) {
+        checkIsRunning();
         send(key, value, null);
     }
 
-    public void send(K key, V value, org.apache.kafka.clients.producer.Callback callback) {
+    public void send(final K key, final V value, final org.apache.kafka.clients.producer.Callback callback) {
+        checkIsRunning();
         producer.send(new ProducerRecord<>(topic, key, value), callback);
     }
 
+    private synchronized void checkIsRunning() {
+        if (state != States.RUNNING) {
+            throw new IllegalStateException("KafkaBasedLog is already not running.");
+        }
+    }
+
 
     private Producer<K, V> createProducer() {
         // Always require producer acks to all to ensure durable writes

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/KafkaStateBackingStore.java

@@ -85,7 +85,7 @@ private synchronized void setState(final States status) {
      * {@inheritDoc}
      */
     @Override
-    public void start() {
+    public synchronized void start() {
         if (isStarted()) {
             throw new IllegalStateException("Cannot init again.");
         }

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/FilePulseSourceTask.java

@@ -23,6 +23,7 @@
 import io.streamthoughts.kafka.connect.filepulse.filter.DefaultRecordFilterPipeline;
 import io.streamthoughts.kafka.connect.filepulse.filter.RecordFilterPipeline;
 import io.streamthoughts.kafka.connect.filepulse.reader.RecordsIterable;
+import io.streamthoughts.kafka.connect.filepulse.state.FileObjectStateBackingStore;
 import io.streamthoughts.kafka.connect.filepulse.state.FileObjectStateBackingStoreManager;
 import io.streamthoughts.kafka.connect.filepulse.state.internal.OpaqueMemoryResource;
 import io.streamthoughts.kafka.connect.filepulse.storage.StateBackingStore;
@@ -189,7 +190,13 @@ private void initSharedStateBackingStore(final String connectorGroupName) {
             sharedStore = FileObjectStateBackingStoreManager.INSTANCE
                     .getOrCreateSharedStore(
                             connectorGroupName,
-                            taskConfig::getStateBackingStore,
+                            () -> {
+                                final FileObjectStateBackingStore store = taskConfig.getStateBackingStore();
+                                // Always invoke the start() method when store is created from Task
+                                // because this means the connector is running on a remote worker.
+                                store.start();
+                                return store;
+                            },
                             new Object()
                     );
         } catch (Exception exception) {

connect-file-pulse-plugin/src/test/java/io/streamthoughts/kafka/connect/filepulse/AbstractKafkaConnectTest.java

@@ -38,14 +38,17 @@
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.containers.wait.Wait;
+import org.testcontainers.containers.wait.strategy.WaitStrategyTarget;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.stream.Stream;
 
+import static org.testcontainers.containers.wait.strategy.Wait.forHttp;
+
 public class AbstractKafkaConnectTest {
 
     private static final Logger LOG = LoggerFactory.getLogger(FilePulseIT.class);
@@ -63,7 +66,7 @@ public class AbstractKafkaConnectTest {
     private static final String CONNECT_PLUGIN_PATH     = "/usr/share/java/";
 
     private static final String DOCKER_USERNAME         = "confluentinc";
-    private static final String DOCKER_CONFLUENT_TAG    = "5.2.1";
+    private static final String DOCKER_CONFLUENT_TAG    = "6.2.0";
     private static final String CP_ZOOKEEPER_IMAGE      = "cp-zookeeper";
     private static final String CP_KAFKA_IMAGE          = "cp-kafka";
     private static final String CP_CONNECT_IMAGE        = "cp-kafka-connect-base";
@@ -137,7 +140,7 @@ private static GenericContainer createConnectWorkerContainer() {
                 .withEnv("CONNECT_REST_ADVERTISED_HOST_NAME", CP_CONNECT_IMAGE)
                 .withEnv("CONNECT_PLUGIN_PATH", CONNECT_PLUGIN_PATH)
                 .withEnv("CONNECT_BOOTSTRAP_SERVERS", KAFKA_NETWORK_ALIAS + ":29092")
-                .waitingFor(Wait.forHttp("/"))
+                .waitingFor(forHttp("/"))
                 .withFileSystemBind(getConnectPluginsDistDir(), CONNECT_PLUGIN_PATH + CONNECTOR_DIR_NAME + "/", BindMode.READ_WRITE);
     }
 
@@ -203,7 +206,7 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
                         }
                     });
             while (true) {
-                ConsumerRecords<String, String> records = consumer.poll(100);
+                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                 records.forEach(System.out::println);
 
             }