fix(plugin): add generation-id prop to task-config to force reconfiguraton

Author: fhussonnois

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

@@ -47,6 +47,9 @@ public class SourceTaskConfig extends CommonSourceConfig {
     private static final String OMIT_READ_COMMITTED_FILE_CONFIG = "ignore.committed.offsets";
     private static final String OMIT_READ_COMMITTED_FILE_DOC = "Should a task ignore committed offsets while scheduling a file (default : false).";
 
+    public static final String TASK_GENERATION_ID = "task.generation.id";
+    private static final String TASK_GENERATION_DOC = "The task configuration generation id.";
+
     private final EnrichedConnectorConfig enrichedConfig;
 
     static ConfigDef getConf() {
@@ -64,6 +67,13 @@ static ConfigDef getConf() {
                         false,
                         ConfigDef.Importance.LOW,
                         OMIT_READ_COMMITTED_FILE_DOC
+                )
+                .define(
+                        TASK_GENERATION_ID,
+                        ConfigDef.Type.INT,
+                        0,
+                        ConfigDef.Importance.LOW,
+                        TASK_GENERATION_DOC
                 );
     }
 
@@ -178,6 +188,10 @@ public FileInputReader reader() {
         return getConfiguredInstance(CommonSourceConfig.TASKS_FILE_READER_CLASS_CONFIG, FileInputReader.class);
     }
 
+    public int getTaskGenerationId() {
+        return this.getInt(TASK_GENERATION_ID);
+    }
+
     public List<RecordFilter> filters() {
         final List<String> filterAliases = getList(FILTER_CONFIG);
         final List<RecordFilter> filters = new ArrayList<>(filterAliases.size());

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

@@ -158,7 +158,7 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
             final List<Map<String, String>> taskConfigs = new ArrayList<>(maxTasks);
             IntStream.range(0, maxTasks)
                     .forEachOrdered(i -> taskConfigs
-                            .add(createTaskConfig(i, maxTasks, null))
+                            .add(createTaskConfig(i, maxTasks, 0, null))
                     );
             return taskConfigs;
         }
@@ -176,12 +176,12 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
             LOG.info("No object file was found - resetting all tasks with an empty config.");
             IntStream.range(0, maxTasks)
                     .forEachOrdered(i -> taskConfigs
-                            .add(createTaskConfig(i, maxTasks, Collections.emptyList()))
+                            .add(createTaskConfig(i, maxTasks, taskConfigsGen, Collections.emptyList()))
                     );
         } else {
             IntStream.range(0, partitioned.size())
                     .forEachOrdered(i -> taskConfigs
-                            .add(createTaskConfig(i, partitioned.size(), partitioned.get(i)))
+                            .add(createTaskConfig(i, partitioned.size(), taskConfigsGen, partitioned.get(i)))
                     );
         }
 
@@ -204,8 +204,10 @@ private List<List<String>> partitionAndGet(int maxTasks) {
 
     private Map<String, String> createTaskConfig(final int taskId,
                                                  final int taskCount,
+                                                 final long taskConfigGen,
                                                  final List<String> URIs) {
         final Map<String, String> taskConfig = new HashMap<>(configProperties);
+        taskConfig.put(SourceTaskConfig.TASK_GENERATION_ID, String.valueOf(taskConfigGen));
         if (connectorConfig.isFileListingTaskDelegationEnabled()) {
             taskConfig.put(SourceTaskConfig.FILE_URIS_PROVIDER_CONFIG, DelegateTaskFileURIProvider.class.getName());
             taskConfig.put(SourceTaskConfig.TASK_PARTITIONER_CLASS_CONFIG, HashByURITaskPartitioner.class.getName());