fix(plugin): fix regression cleanup object files should not be rescheduled (#178)

Author: fhussonnois

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

@@ -60,11 +60,15 @@ public enum FileObjectStatus {
     FAILED,
 
     /**
-     * The file has been successfully clean up (depending of the configured strategy).
+     * The file has been successfully clean up (depending on the configured strategy).
      */
     CLEANED;
 
     public boolean isOneOf(final FileObjectStatus...states) {
         return Arrays.asList(states).contains(this);
     }
+
+    public boolean isDone() {
+        return isOneOf(COMMITTED, FAILED, CLEANED);
+    }
 }

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

@@ -289,7 +289,10 @@ private synchronized boolean updateFiles() {
                 offsetPolicy,
                 fileObjectKey -> {
                     final FileObject fileObject = snapshot.getForKey(fileObjectKey.original());
-                    return !(fileObject != null && cleanablePredicate.test(fileObject.status()));
+                    if (fileObject == null) return true;
+
+                    final FileObjectStatus status = fileObject.status();
+                    return !(cleanablePredicate.test(status) || status.isDone());
                 },
                 objects
         );
@@ -378,6 +381,7 @@ public List<FileObjectMeta> listFilesToSchedule(final int maxFilesToSchedule) {
                         LOG.warn("Failed to read state changelog while scheduling object files. Timeout.");
                     }
 
+                    // Check if all scanned object-files can be schedule.
                     if (scanned.size() <= maxFilesToSchedule) {
                         scheduled.putAll(scanned);
                     } else {
@@ -388,10 +392,7 @@ public List<FileObjectMeta> listFilesToSchedule(final int maxFilesToSchedule) {
                         }
                     }
 
-                    partitions = scanned.values()
-                            .stream()
-                            .sorted(BY_LAST_MODIFIED)
-                            .collect(Collectors.toList());
+                    partitions = new ArrayList<>(scheduled.values());
 
                     attempts++;
                     if (changed.get()) {
@@ -411,10 +412,17 @@ public List<FileObjectMeta> listFilesToSchedule(final int maxFilesToSchedule) {
             }
 
             if (partitions.isEmpty()) {
-                LOG.warn("Filesystem could not be scanned quickly enough, " +
-                        "or no object file was detected after starting the connector.");
+                LOG.warn(
+                    "Filesystem could not be scanned quickly enough, " +
+                    "or no object file was detected after starting the connector."
+                );
             }
-            return partitions;
+
+            // Sort all object files by the last-modified date before returning
+            return partitions
+                    .stream()
+                    .sorted(BY_LAST_MODIFIED)
+                    .collect(Collectors.toList());
         } finally {
             scanned.clear();
             taskReconfigurationRequested.set(false);

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

@@ -21,6 +21,7 @@
 import io.streamthoughts.kafka.connect.filepulse.config.CommonSourceConfig;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObject;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObjectStatus;
 import io.streamthoughts.kafka.connect.filepulse.source.SourceOffsetPolicy;
 import io.streamthoughts.kafka.connect.filepulse.source.TaskPartitioner;
 import io.streamthoughts.kafka.connect.filepulse.state.StateBackingStoreAccess;
@@ -95,16 +96,15 @@ public List<URI> nextURIs() {
                     if (fileObject == null)
                         return true;
 
-                    switch (fileObject.status()) {
-                        case COMMITTED:
-                        case FAILED:
-                        case CLEANED:
-                            return false;
-                        case COMPLETED:
-                            return isFirstCall;
-                        default:
-                            return true;
-                    }
+                    final FileObjectStatus status = fileObject.status();
+
+                    // If an object file is marked as COMPLETED (i.e. not COMMITTED) we should only consider
+                    // it processable the first time this method is called.
+                    if (status == FileObjectStatus.COMPLETED)
+                        return isFirstCall;
+
+                    // Otherwise, only return true if the is file not already done.DelegateTaskFileURIProvider
+                    return !status.isDone();
                 },
                 fileSystemListing.listObjects()
         ).values();

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

@@ -36,7 +36,7 @@
 import java.util.stream.Stream;
 
 /**
- * Class which is used to determinate if a a list of file can be processed.
+ * Class which is used to determinate if a list of file can be processed.
  */
 public class FileObjectCandidatesFilter {
 
@@ -70,7 +70,7 @@ public Map<FileObjectKey, FileObjectMeta> filter(final Collection<FileObjectMeta
                 .filter(kv -> predicate.test(FileObjectKey.of(kv.key)))
                 .collect(Collectors.toList());
 
-        // Looking for duplicates in sources files, i.e the OffsetPolicy generate two identical offsets for two files.
+        // Looking for duplicates in object files, i.e., the OffsetPolicy generates two identical offsets for two files.
         final Stream<Map.Entry<String, List<KeyValuePair<String, FileObjectMeta>>>> entryStream = toScheduled
                 .stream()
                 .collect(Collectors.groupingBy(kv -> kv.key))