feat(plugin): add new config tasks.empty.poll.wait.ms

This commit adds the new prop `tasks.empty.poll.wait.ms` to fix the amount of time in millisecond
a tasks should wait if a poll returns an empty list of records.

Author: fhussonnois

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

@@ -55,6 +55,9 @@ public class CommonSourceConfig extends AbstractConfig {
     public static final String TASKS_HALT_ON_ERROR_CONFIG     = "tasks.halt.on.error";
     private static final String TASKS_HALT_ON_ERROR_DOC       = "Should a task halt when it encounters an error or continue to the next file.";
 
+    public static final String TASKS_EMPTY_POLL_WAIT_MS_CONFIG = "tasks.empty.poll.wait.ms";
+    public static final String TASKS_EMPTY_POLL_WAIT_MS_DOC    = "The amount of time in millisecond a tasks should wait if a poll returns an empty list of records.";
+
     public static final String OFFSET_STRATEGY_CLASS_CONFIG = "offset.policy.class";
     private static final String OFFSET_STRATEGY_CLASS_DOC = "Class which is used to determine the source partition and offset that uniquely identify a input record";
     private static final String OFFSET_STRATEGY_CLASS_DEFAULT = DefaultSourceOffsetPolicy.class.getName();
@@ -114,6 +117,17 @@ public static ConfigDef getConfigDev() {
                         ConfigDef.Width.NONE,
                         TASKS_HALT_ON_ERROR_CONFIG
                 )
+                .define(
+                        TASKS_EMPTY_POLL_WAIT_MS_CONFIG,
+                        ConfigDef.Type.LONG,
+                        500,
+                        ConfigDef.Importance.LOW,
+                        TASKS_EMPTY_POLL_WAIT_MS_DOC,
+                        GROUP,
+                        groupCounter++,
+                        ConfigDef.Width.NONE,
+                        TASKS_EMPTY_POLL_WAIT_MS_CONFIG
+                )
                 .define(
                         OUTPUT_TOPIC_CONFIG,
                         ConfigDef.Type.STRING,
@@ -169,7 +183,7 @@ public static ConfigDef getConfigDev() {
                         TASK_PARTITIONER_CLASS_DOC
                 );
     }
-    
+
     public FileSystemListing<?> getFileSystemListing() {
         return getConfiguredInstance(FS_LISTING_CLASS_CONFIG, FileSystemListing.class);
     }
@@ -186,6 +200,10 @@ public boolean isTaskHaltOnError() {
         return this.getBoolean(TASKS_HALT_ON_ERROR_CONFIG);
     }
 
+    public long getTaskEmptyPollWaitMs() {
+        return this.getLong(TASKS_EMPTY_POLL_WAIT_MS_CONFIG);
+    }
+
     public SourceOffsetPolicy getSourceOffsetPolicy() {
         return this.getConfiguredInstance(OFFSET_STRATEGY_CLASS_CONFIG, SourceOffsetPolicy.class);
     }

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

@@ -51,8 +51,6 @@ public class FilePulseSourceTask extends SourceTask {
 
     private static final Integer NO_PARTITION = null;
 
-    private static final int DEFAULT_POLL_WAIT_MS = 500;
-
     public SourceTaskConfig taskConfig;
 
     private String topic;
@@ -216,8 +214,8 @@ public List<SourceRecord> poll() throws InterruptedException {
     }
 
     private void busyWait() throws InterruptedException {
-        LOG.trace("Waiting {} ms to poll next records", DEFAULT_POLL_WAIT_MS);
-        Thread.sleep(DEFAULT_POLL_WAIT_MS);
+        LOG.trace("Waiting {} ms to poll next records", taskConfig.getTaskEmptyPollWaitMs());
+        Thread.sleep(taskConfig.getTaskEmptyPollWaitMs());
     }
 
     /**