fix(plugin): fix NPE from KafkaBasedLog when using mutliple worker

Author: fhussonnois

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

@@ -97,10 +97,9 @@ public void start(final Map<String, String> props) {
         final String tasksReporterGroupId = config.getTasksReporterGroupId();
         connectorGroupName = tasksReporterGroupId != null ? tasksReporterGroupId : connectName;
         StateBackingStoreRegistry.instance().register(connectorGroupName, () -> {
+            final String stateStoreTopic = config.getTaskReporterTopic();
             final Map<String, Object> configs = config.getInternalKafkaReporterConfig();
-            return new FileStateBackingStore(
-                    config.getTaskReporterTopic(),
-                    connectorGroupName, configs);
+            return new FileStateBackingStore(stateStoreTopic, connectorGroupName, configs, false);
         });
 
         final FSDirectoryWalker directoryScanner = this.config.directoryScanner();

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

@@ -99,9 +99,10 @@ private StateBackingStore<SourceFile> getStateStatesBackingStore() {
         final String groupId = config.getTasksReporterGroupId();
         StateBackingStoreRegistry.instance().register(groupId, () -> {
             final Map<String, Object> configs = config.getInternalKafkaReporterConfig();
-            return new FileStateBackingStore(
-                    config.getTaskReporterTopic(),
-                    groupId, configs);
+            final String stateStoreTopic = config.getTaskReporterTopic();
+            FileStateBackingStore store = new FileStateBackingStore(stateStoreTopic, groupId, configs, true);
+            store.start();
+            return store;
         });
 
         return StateBackingStoreRegistry.instance().get(groupId);

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

@@ -53,6 +53,7 @@ public class KafkaFileStateReporter implements StateListener {
 
     /**
      * Notify a state change for the specified source file.
+     *
      * @param metadata  the source file metadata.
      * @param offset    the source file offset.
      * @param status    the status.

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/state/FileStateBackingStore.java

@@ -32,13 +32,15 @@ public class FileStateBackingStore extends KafkaStateBackingStore<SourceFile> {
     /**
      * Creates a new {@link FileStateBackingStore} instance.
      *
-     * @param store   the state store name.
-     * @param groupId the group attached to the backing store.
-     * @param configs the configuration.
+     * @param store          the state store name.
+     * @param groupId        the group attached to the backing store.
+     * @param configs        the configuration.
+     * @param isProducerOnly is the backing store only used for writing data.
      */
     public FileStateBackingStore(final String store,
                                  final String groupId,
-                                 final Map<String, ?> configs) {
-        super(store, KEY_PREFIX, groupId, configs, new SourceFileSerde());
+                                 final Map<String, ?> configs,
+                                 final boolean isProducerOnly) {
+        super(store, KEY_PREFIX, groupId, configs, new SourceFileSerde(), isProducerOnly);
     }
 }

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

@@ -73,7 +73,7 @@ enum States {
 
     /**
      * Create a new KafkaBasedLog object. This does not start reading the log and writing is not permitted until
-     * {@link #start()} is invoked.
+     * {@link #start(boolean)} is invoked.
      *
      * @param topic the topic to treat as a log
      * @param producerConfigs configuration options to use when creating the internal producer. At a minimum this must
@@ -86,7 +86,7 @@ enum States {
      *                        behavior of this class.
      * @param consumedCallback callback to invoke for each {@link ConsumerRecord} consumed when tailing the log
      * @param time Time interface
-     * @param initializer the component that should be run when this log is {@link #start() started}; may be null
+     * @param initializer the component that should be run when this log is {@link #start(boolean) started}; may be null
      */
     KafkaBasedLog(final String topic,
                   final Map<String, Object> producerConfigs,
@@ -105,43 +105,44 @@ enum States {
         this.state = States.CREATED;
     }
 
-    public synchronized void start() {
+    public synchronized void start(final boolean producerOnly) {
         if (state != States.CREATED) {
             throw new IllegalStateException("Cannot restart KafkaBasedLog due to state being " + state +")");
         }
         LOG.info("Starting KafkaBasedLog with topic {}",  topic);
         try {
             initializer.run();
             producer = createProducer();
-            consumer = createConsumer();
+            if (!producerOnly) {
+                consumer = createConsumer();
 
-            List<TopicPartition> partitions = new ArrayList<>();
+                List<TopicPartition> partitions = new ArrayList<>();
 
-            // We expect that the topics will have been created either manually by the user or automatically by the herder
-            List<PartitionInfo> partitionInfos = null;
-            long started = time.milliseconds();
-            while (partitionInfos == null && time.milliseconds() - started < CREATE_TOPIC_TIMEOUT_MS) {
-                partitionInfos = consumer.partitionsFor(topic);
-                Utils.sleep(Math.min(time.milliseconds() - started, 1000));
-            }
-            if (partitionInfos == null)
-                throw new ConnectException(
-                    "Could not look up partition metadata for position backing store topic '" + topic + "' in" +
-                    " allotted period (" + CREATE_TOPIC_TIMEOUT_MS + "ms). This could indicate a connectivity issue," +
-                    " unavailable topic partitions, or if this is your first use of the topic it may have taken too long to create.");
+                List<PartitionInfo> partitionInfos = null;
+                long started = time.milliseconds();
+                while (partitionInfos == null && time.milliseconds() - started < CREATE_TOPIC_TIMEOUT_MS) {
+                    partitionInfos = consumer.partitionsFor(topic);
+                    Utils.sleep(Math.min(time.milliseconds() - started, 1000));
+                }
+                if (partitionInfos == null)
+                    throw new ConnectException(
+                        "Could not look up partition metadata for position backing store topic '" + topic + "' in" +
+                        " allotted period (" + CREATE_TOPIC_TIMEOUT_MS + "ms). This could indicate a connectivity issue," +
+                        " unavailable topic partitions, or if this is your first use of the topic it may have taken too long to create.");
 
-            for (PartitionInfo partition : partitionInfos)
-                partitions.add(new TopicPartition(partition.topic(), partition.partition()));
-            consumer.assign(partitions);
+                for (PartitionInfo partition : partitionInfos)
+                    partitions.add(new TopicPartition(partition.topic(), partition.partition()));
+                consumer.assign(partitions);
 
-            // Always consume from the beginning of all partitions. Necessary to ensure that we don't use committed offsets
-            // when a 'group.id' is specified (if offsets happen to have been committed unexpectedly).
-            consumer.seekToBeginning(partitions);
+                // Always consume from the beginning of all partitions. Necessary to ensure that we don't use committed offsets
+                // when a 'group.id' is specified (if offsets happen to have been committed unexpectedly).
+                consumer.seekToBeginning(partitions);
 
-            readToLogEnd();
+                readToLogEnd();
 
-            thread = new WorkThread();
-            thread.start();
+                thread = new WorkThread();
+                thread.start();
+            }
             state = States.RUNNING;
             LOG.info("Finished reading KafkaBasedLog for topic {}", topic);
             LOG.info("Started KafkaBasedLog for topic {}", topic);

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

@@ -46,6 +46,7 @@ public class KafkaStateBackingStore<T> implements StateBackingStore<T> {
     private final Map<String, T> states = new HashMap<>();
     private final StateSerde<T> serde;
     private final String keyPrefix;
+    private final boolean isProducerOnly;
     private States status = States.CREATED;
     private StateBackingStore.UpdateListener<T> updateListener;
 
@@ -62,12 +63,14 @@ public KafkaStateBackingStore(final String topic,
                                   final String keyPrefix,
                                   final String groupId,
                                   final Map<String, ?> configs,
-                                  final StateSerde<T> serde) {
+                                  final StateSerde<T> serde,
+                                  final boolean isProducerOnly) {
         KafkaBasedLogFactory factory = new KafkaBasedLogFactory(configs);
         this.configLog = factory.make(topic, new ConsumeCallback());
         this.groupId = groupId;
         this.serde = serde;
         this.keyPrefix = keyPrefix;
+        this.isProducerOnly = isProducerOnly;
     }
 
     synchronized States getState() {
@@ -89,7 +92,7 @@ public void start() {
         LOG.info("Starting {}", getBackingStoreName());
         // Before startup, callbacks are *not* invoked. You can grab a snapshot after starting -- just take care that
         // updates can continue to occur in the background
-        configLog.start();
+        configLog.start(isProducerOnly);
         setState(States.STARTED);
         LOG.info("Started {}", getBackingStoreName());
     }