feat(plugin): add new property max.scheduled.files (#122) (#123)

Resolves: #122
Resolves: #123

Author: fhussonnois

checkstyle/suppressions.xml

@@ -42,7 +42,7 @@
  -->
     <suppress checks="LineLength" files="RecordFilter.java"/>
     <suppress checks="LineLength" files="GrokPatternCompiler.java"/>
-    <suppress checks="LineLength" files="DefaultFileSystemScanner.java"/>
+    <suppress checks="LineLength" files="DefaultFileSystemMonitor.java"/>
     <suppress checks="LineLength" files="ComposeOffsetStrategy.java"/>
     <suppress checks="LineLength" files=".*Config.java"/>
     <!-- Those classes are copy from kafka-connect api-->

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

@@ -24,24 +24,34 @@
 import java.util.List;
 
 /**
- * A {@link FileSystemScanner} is responsible to scan a specific file system
- * for new files to stream into Kafka;
+ * A {@code FileSystemMonitor} is responsible for scanning a specific file system for new files to stream into Kafka.
  */
-public interface FileSystemScanner {
+public interface FileSystemMonitor {
 
     /**
-     * Run a single file system scan using the specified context.
+     * Run a single filesystem scan using the specified context.
      * @param context   the connector context.
      */
-    void scan(final ConnectorContext context);
+    void invoke(final ConnectorContext context);
 
     /**
      * Gets newest files found during last scan partitioned for the specified number of groups.
      *
-     * @param maxGroups the maximum number of groups.
+     * @param maxGroups          the maximum number of groups.
+     * @return                   list of files to execute.
+     */
+   default List<List<URI>> partitionFilesAndGet(final int maxGroups) {
+       return partitionFilesAndGet(maxGroups, Integer.MAX_VALUE);
+   }
+
+    /**
+     * Gets newest files found during last scan partitioned for the specified number of groups.
+     *
+     * @param maxGroups          the maximum number of groups.
+     * @param maxFilesToSchedule the maximum number of files to scheduled.
      * @return          list of files to execute.
      */
-    List<List<URI>> partitionFilesAndGet(final int maxGroups);
+    List<List<URI>> partitionFilesAndGet(final int maxGroups, int maxFilesToSchedule);
 
     /**
      * Close underlying I/O resources.

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

@@ -37,6 +37,10 @@ public class ConnectorConfig extends CommonConfig {
     public static final String FS_LISTING_FILTERS_CONFIG      = "fs.listing.filters";
     private static final String FS_SCAN_FILTERS_DOC           = "Filters classes which are used to apply list input files.";
 
+    public static final String MAX_SCHEDULED_FILES_CONFIG     = "max.scheduled.files";
+    public static final String MAX_SCHEDULED_FILES_DOC        = "Maximum number of files that can be schedules to tasks.";
+    public static final int MAX_SCHEDULED_FILES_DEFAULT       = 1000;
+
     public static final String ALLOW_TASKS_RECONFIG_AFTER_TIMEOUT_MS_CONFIG = "allow.tasks.reconfiguration.after.timeout.ms";
     public static final String ALLOW_TASKS_RECONFIG_AFTER_TIMEOUT_MS_DOC = "Specifies the timeout (in milliseconds) for the connector to allow tasks to be reconfigured when new files are detected, even if some tasks are still being processed.";
 
@@ -65,23 +69,60 @@ public ConnectorConfig(final Map<?, ?> originals) {
 
     public static ConfigDef getConf() {
         return CommonConfig.getConf()
-                .define(FS_LISTING_CLASS_CONFIG, ConfigDef.Type.CLASS,
-                        LocalFSDirectoryListing.class, ConfigDef.Importance.HIGH, FS_LISTING_CLASS_DOC)
-
-                .define(FS_LISTING_FILTERS_CONFIG, ConfigDef.Type.LIST, Collections.emptyList(),
-                        ConfigDef.Importance.MEDIUM, FS_SCAN_FILTERS_DOC)
-
-                .define(FS_SCAN_INTERVAL_MS_CONFIG, ConfigDef.Type.LONG, FS_SCAN_INTERVAL_MS_DEFAULT,
-                        ConfigDef.Importance.HIGH, FS_SCAN_INTERVAL_MS_DOC)
-
-                .define(FILE_CLEANER_CLASS_CONFIG,
-                        ConfigDef.Type.CLASS, ConfigDef.Importance.HIGH, FILE_CLEANER_CLASS_DOC)
-
-                .define(ALLOW_TASKS_RECONFIG_AFTER_TIMEOUT_MS_CONFIG, ConfigDef.Type.LONG, Long.MAX_VALUE,
-                        ConfigDef.Importance.MEDIUM, ALLOW_TASKS_RECONFIG_AFTER_TIMEOUT_MS_DOC)
-
-                .define(INTERNAL_REPORTER_GROUP_ID, ConfigDef.Type.STRING, null,
-                        ConfigDef.Importance.MEDIUM, INTERNAL_REPORTER_GROUP_ID_DOC);
+                .define(
+                        FS_LISTING_CLASS_CONFIG,
+                        ConfigDef.Type.CLASS,
+                        LocalFSDirectoryListing.class,
+                        ConfigDef.Importance.HIGH,
+                        FS_LISTING_CLASS_DOC
+                )
+
+                .define(
+                        FS_LISTING_FILTERS_CONFIG,
+                        ConfigDef.Type.LIST,
+                        Collections.emptyList(),
+                        ConfigDef.Importance.MEDIUM,
+                        FS_SCAN_FILTERS_DOC
+                )
+
+                .define(
+                        FS_SCAN_INTERVAL_MS_CONFIG,
+                        ConfigDef.Type.LONG,
+                        FS_SCAN_INTERVAL_MS_DEFAULT,
+                        ConfigDef.Importance.HIGH,
+                        FS_SCAN_INTERVAL_MS_DOC
+                )
+
+                .define(
+                        FILE_CLEANER_CLASS_CONFIG,
+                        ConfigDef.Type.CLASS,
+                        ConfigDef.Importance.HIGH,
+                        FILE_CLEANER_CLASS_DOC
+                )
+
+                .define(
+                        ALLOW_TASKS_RECONFIG_AFTER_TIMEOUT_MS_CONFIG,
+                        ConfigDef.Type.LONG,
+                        Long.MAX_VALUE,
+                        ConfigDef.Importance.MEDIUM,
+                        ALLOW_TASKS_RECONFIG_AFTER_TIMEOUT_MS_DOC
+                )
+
+                .define(
+                        INTERNAL_REPORTER_GROUP_ID,
+                        ConfigDef.Type.STRING,
+                        null,
+                        ConfigDef.Importance.MEDIUM,
+                        INTERNAL_REPORTER_GROUP_ID_DOC
+                )
+
+                .define(
+                        MAX_SCHEDULED_FILES_CONFIG,
+                        ConfigDef.Type.INT,
+                        MAX_SCHEDULED_FILES_DEFAULT,
+                        ConfigDef.Importance.MEDIUM,
+                        MAX_SCHEDULED_FILES_DOC
+                );
     }
 
     public Long allowTasksReconfigurationAfterTimeoutMs() {
@@ -92,6 +133,10 @@ public String getTasksReporterGroupId() {
         return getString(INTERNAL_REPORTER_GROUP_ID);
     }
 
+    public int getMaxScheduledFiles() {
+        return getInt(MAX_SCHEDULED_FILES_CONFIG);
+    }
+
     public FileCleanupPolicy cleanupPolicy() {
         return getConfiguredInstance(FILE_CLEANER_CLASS_CONFIG, FileCleanupPolicy.class);
     }