feat(plugin): add support for task file listing delegation

This commit adds the ability to delegate the file listing process to tasks.
This new capability is intended for use in contexts where a large number of
files are created in very short periods on the local file system.

Changes:
- add new config property fs.listing.tasks.delegation.enabled

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/FileSystemMonitor.java

@@ -37,6 +37,14 @@ public interface FileSystemMonitor {
      */
     void invoke(final ConnectorContext context);
 
+    /**
+     * Enables or disables the the file-listing process either temporarily or permanently.
+     * In other words, if disabled then {@link #listFilesToSchedule()} will always return an empty list.
+     *
+     * @param enabled is the file-listing process enabled.
+     */
+    void setFileSystemListingEnabled(final boolean enabled);
+
     /**
      * Retrieves the list of objects-files that were found during the last the {@link #invoke(ConnectorContext)} call.
      * This method should not return more than the given maximum.

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/FileRecordsPollingConsumer.java

@@ -39,5 +39,5 @@ public interface FileRecordsPollingConsumer<T> extends FileInputIterator<T> {
      *
      * @param listener  the {@link StateListener} instance to be used.
      */
-    void setFileListener(final StateListener listener);
+    void setStateListener(final StateListener listener);
 }

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/StateListener.java

@@ -18,7 +18,7 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.source;
 
-interface StateListener {
+public interface StateListener {
 
     /**
      * This method is invoked when a file is scheduled by the task.

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/TaskPartitioner.java

@@ -19,9 +19,11 @@
 package io.streamthoughts.kafka.connect.filepulse.source;
 
 import java.net.URI;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 
-public interface TaskPartitioner {
+public interface TaskPartitioner extends AutoCloseable {
 
     /**
      * Partitions the specified object-file URIs.
@@ -30,7 +32,7 @@ public interface TaskPartitioner {
      * @param taskCount     the total number of tasks.
      * @return              the list of URIs for the given {@literal taskId}.
      */
-    List<List<URI>> partition(final List<FileObjectMeta> files, final int taskCount);
+    List<List<URI>> partition(final Collection<FileObjectMeta> files, final int taskCount);
 
     /**
      * Partitions the specified object-file URIs.
@@ -40,9 +42,15 @@ public interface TaskPartitioner {
      * @param taskId        the task id.
      * @return              the list of URIs for the given {@literal taskId}.
      */
-    default List<URI> partitionForTask(final List<FileObjectMeta> files,
+    default List<URI> partitionForTask(final Collection<FileObjectMeta> files,
                                        final int taskCount,
                                        final int taskId) {
-        return partition(files, taskCount).get(taskId);
+        return files == null || files.isEmpty() ? Collections.emptyList() : partition(files, taskCount).get(taskId);
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    default void close() { }
 }

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/SourceConnectorConfig.java

@@ -42,6 +42,9 @@ public class SourceConnectorConfig extends CommonSourceConfig {
     private static final String MAX_SCHEDULED_FILES_DOC       = "Maximum number of files that can be schedules to tasks.";
     private static final int MAX_SCHEDULED_FILES_DEFAULT      = 1000;
 
+    public static final String FS_LISTING_TASK_DELEGATION_ENABLED_CONFIG = "fs.listing.task.delegation.enabled";
+    private static final String FS_LISTING_TASK_DELEGATION_ENABLED_DOC = "Boolean indicating whether the file listing process should be delegated to tasks.";
+
     /**
      * Creates a new {@link SourceConnectorConfig} instance.
      * @param originals the originals configuration.
@@ -81,6 +84,14 @@ public static ConfigDef getConf() {
                         MAX_SCHEDULED_FILES_DEFAULT,
                         ConfigDef.Importance.MEDIUM,
                         MAX_SCHEDULED_FILES_DOC
+                )
+
+                .define(
+                        FS_LISTING_TASK_DELEGATION_ENABLED_CONFIG,
+                        ConfigDef.Type.BOOLEAN,
+                        false,
+                        ConfigDef.Importance.LOW,
+                        FS_LISTING_TASK_DELEGATION_ENABLED_DOC
                 );
     }
 
@@ -92,12 +103,16 @@ public int getMaxScheduledFiles() {
         return getInt(MAX_SCHEDULED_FILES_CONFIG);
     }
 
-    public FileCleanupPolicy cleanupPolicy() {
+    public FileCleanupPolicy getFileCleanupPolicy() {
         return getConfiguredInstance(FILE_CLEANER_CLASS_CONFIG, FileCleanupPolicy.class);
     }
 
-    public long scanInternalMs() {
+    public long getListingInterval() {
         return this.getLong(FS_LISTING_INTERVAL_MS_CONFIG);
     }
 
+    public boolean isFileListingTaskDelegationEnabled() {
+        return getBoolean(FS_LISTING_TASK_DELEGATION_ENABLED_CONFIG);
+    }
+
 }

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/SourceTaskConfig.java

@@ -41,7 +41,7 @@
  */
 public class SourceTaskConfig extends CommonSourceConfig {
 
-    private static final String FILE_URIS_PROVIDER_CONFIG = "file.uris.provider";
+    public static final String FILE_URIS_PROVIDER_CONFIG = "file.uris.provider";
     private static final String FILE_URIS_PROVIDER_DOC    = "The FileURIProvider class to be used for retrieving the file URIs to process";
 
     private static final String OMIT_READ_COMMITTED_FILE_CONFIG = "ignore.committed.offsets";

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/DefaultFileSystemMonitor.java

@@ -24,8 +24,6 @@
 import io.streamthoughts.kafka.connect.filepulse.clean.FileCleanupPolicyResult;
 import io.streamthoughts.kafka.connect.filepulse.clean.FileCleanupPolicyResultSet;
 import io.streamthoughts.kafka.connect.filepulse.clean.GenericFileCleanupPolicy;
-import io.streamthoughts.kafka.connect.filepulse.config.SourceConnectorConfig;
-import io.streamthoughts.kafka.connect.filepulse.internal.KeyValuePair;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObject;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObjectKey;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
@@ -54,7 +52,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 /**
  * A default {@link FileSystemMonitor} that can be used to trigger file
@@ -102,6 +99,8 @@ public class DefaultFileSystemMonitor implements FileSystemMonitor {
 
     private final AtomicBoolean changed = new AtomicBoolean(false);
 
+    private final AtomicBoolean fileSystemListingEnabled = new AtomicBoolean(true);
+
     /**
      * Creates a new {@link DefaultFileSystemMonitor} instance.
      *
@@ -211,7 +210,7 @@ private boolean readStatesToEnd(final Duration timeout) {
     public void invoke(final ConnectorContext context) {
         // It seems to be OK to always run cleanup even if connector is not yet started or is being shut down.
         cleanUpCompletedFiles();
-        if (running.get()) {
+        if (running.get() && fileSystemListingEnabled.get()) {
             if (!taskReconfigurationRequested.get()) {
                 if (updateFiles()) {
                     LOG.info("Requesting task reconfiguration");
@@ -221,11 +220,19 @@ public void invoke(final ConnectorContext context) {
             } else {
                 LOG.info("Task reconfiguration requested. Skip filesystem listing.");
             }
-        } else {
+        } else if (fileSystemListingEnabled.get()) {
             LOG.info("The connector is not completely started or is being shut down. Skip filesystem listing.");
         }
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void setFileSystemListingEnabled(final boolean enabled) {
+        this.fileSystemListingEnabled.set(enabled);
+    }
+
     private void cleanUpCompletedFiles() {
         if (completed.isEmpty()) {
             LOG.debug("Skipped cleanup. No object file completed.");
@@ -271,8 +278,8 @@ private synchronized boolean updateFiles() {
         long took = Time.SYSTEM.milliseconds() - started;
         LOG.info("Completed object files listing. '{}' object files found in {}ms", objects.size(), took);
 
-        final StateSnapshot<FileObject> snapshot = store.snapshot();
-        final Map<FileObjectKey, FileObjectMeta> toScheduled = toScheduled(objects, snapshot);
+        final Map<FileObjectKey, FileObjectMeta> toScheduled = FileObjectCandidatesFilter
+                .filter(offsetPolicy, store.snapshot(), objects);
 
         // Some scheduled files are still being processed, but new files are detected
         if (!noScheduledFiles) {
@@ -310,52 +317,6 @@ private synchronized boolean updateFiles() {
         return !scanned.isEmpty() && running.get();
     }
 
-    private Map<FileObjectKey, FileObjectMeta> toScheduled(final Collection<FileObjectMeta> scanned,
-                                                           final StateSnapshot<FileObject> snapshot) {
-
-        final List<KeyValuePair<String, FileObjectMeta>> toScheduled = scanned.stream()
-                .map(source -> KeyValuePair.of(offsetPolicy.toPartitionJson(source), source))
-                .filter(kv -> maybeScheduled(snapshot, kv.key))
-                .collect(Collectors.toList());
-
-        // Looking for duplicates in sources files, i.e the OffsetPolicy generate two identical offsets for two files.
-        final Stream<Map.Entry<String, List<KeyValuePair<String, FileObjectMeta>>>> entryStream = toScheduled
-                .stream()
-                .collect(Collectors.groupingBy(kv -> kv.key))
-                .entrySet()
-                .stream()
-                .filter(entry -> entry.getValue().size() > 1);
-
-        final Map<String, List<String>> duplicates = entryStream
-                .collect(Collectors.toMap(
-                        Map.Entry::getKey,
-                        e -> e.getValue().stream().map(m -> m.value.stringURI()).collect(Collectors.toList()))
-                );
-
-        if (!duplicates.isEmpty()) {
-            final String formatted = duplicates
-                    .entrySet()
-                    .stream()
-                    .map(e -> "partition_key=" + e.getKey() + ", files=" + e.getValue())
-                    .collect(Collectors.joining("\n\t", "\n\t", "\n"));
-            LOG.error(
-                    "Duplicates object files detected. " +
-                    "Consider changing the configuration for '{}'. " +
-                    "Scan is ignored: {}",
-                    SourceConnectorConfig.OFFSET_STRATEGY_CLASS_CONFIG,
-                    formatted
-            );
-            return Collections.emptyMap(); // ignore all sources files
-        }
-
-        return toScheduled.stream().collect(Collectors.toMap(kv -> FileObjectKey.of(kv.key), kv -> kv.value));
-    }
-
-    private boolean maybeScheduled(final StateSnapshot<FileObject> snapshot,
-                                   final String partition) {
-        return !snapshot.contains(partition) ||
-                snapshot.getForKey(partition).status().isOneOf(FileObjectStatus.started());
-    }
 
     /**
      * {@inheritDoc}