feat: add new config props to define file ordering

Author: fhussonnois

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

@@ -21,6 +21,7 @@
 import com.jsoniter.JsonIterator;
 import io.streamthoughts.kafka.connect.filepulse.fs.FileListFilter;
 import io.streamthoughts.kafka.connect.filepulse.fs.FileSystemListing;
+import io.streamthoughts.kafka.connect.filepulse.fs.TaskFileOrder;
 import io.streamthoughts.kafka.connect.filepulse.internal.StringUtils;
 import io.streamthoughts.kafka.connect.filepulse.offset.DefaultSourceOffsetPolicy;
 import io.streamthoughts.kafka.connect.filepulse.source.DefaultTaskPartitioner;
@@ -59,6 +60,9 @@ public class CommonSourceConfig extends AbstractConfig {
     public static final String TASKS_FILE_READER_CLASS_CONFIG = "tasks.reader.class";
     private static final String TASKS_FILE_READER_CLASS_DOC   = "Class which is used by tasks to read an input file.";
 
+    public static final String TASKS_FILE_PROCESSING_ORDER_BY_CONFIG = "tasks.file.processing.order.by";
+    private static final String TASKS_FILE_PROCESSING_ORDER_BY_DOC = "The strategy to be used for sorting files for processing. Valid values are: LAST_MODIFIED, URI, CONTENT_LENGTH, CONTENT_LENGTH_DESC.";
+
     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.";
 
@@ -204,7 +208,7 @@ public static ConfigDef getConfigDev() {
                         ConfigDef.Importance.HIGH,
                         FILTER_DOC,
                         FILTERS_GROUP,
-                        -1,
+                        groupCounter++,
                         ConfigDef.Width.NONE,
                         FILTER_CONFIG
                 )
@@ -214,10 +218,21 @@ public static ConfigDef getConfigDev() {
                         DefaultTaskPartitioner.class,
                         ConfigDef.Importance.HIGH,
                         FILTER_DOC,
+                        TASK_PARTITIONER_CLASS_DOC,
+                        groupCounter++,
+                        ConfigDef.Width.NONE,
+                        TASK_PARTITIONER_CLASS_CONFIG
+                )
+                .define(
+                        TASKS_FILE_PROCESSING_ORDER_BY_CONFIG,
+                        ConfigDef.Type.STRING,
+                        TaskFileOrder.BuiltIn.LAST_MODIFIED.name(),
+                        ConfigDef.Importance.MEDIUM,
+                        TASKS_FILE_PROCESSING_ORDER_BY_DOC,
                         FILTERS_GROUP,
-                        -1,
+                        groupCounter++,
                         ConfigDef.Width.NONE,
-                        TASK_PARTITIONER_CLASS_DOC
+                        TASKS_FILE_PROCESSING_ORDER_BY_CONFIG
                 );
     }
 
@@ -233,6 +248,10 @@ public TaskPartitioner getTaskPartitioner() {
         return this.getConfiguredInstance(TASK_PARTITIONER_CLASS_CONFIG, TaskPartitioner.class);
     }
 
+    public TaskFileOrder getTaskFilerOrder() {
+        return TaskFileOrder.findBuiltInByName(this.getString(TASKS_FILE_PROCESSING_ORDER_BY_CONFIG));
+    }
+
     public boolean isTaskHaltOnError() {
         return this.getBoolean(TASKS_HALT_ON_ERROR_CONFIG);
     }

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

@@ -40,7 +40,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Comparator;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -52,7 +51,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
-import java.util.stream.Collectors;
 
 /**
  * A default {@link FileSystemMonitor} that can be used to trigger file
@@ -68,9 +66,6 @@ public class DefaultFileSystemMonitor implements FileSystemMonitor {
     private static final Duration DEFAULT_READ_END_LOG_TIMEOUT = Duration.ofSeconds(5);
     private static final int MAX_SCHEDULE_ATTEMPTS = 3;
 
-    private final static Comparator<FileObjectMeta> BY_LAST_MODIFIED =
-            Comparator.comparingLong(FileObjectMeta::lastModified);
-
     private final FileSystemListing<?> fsListing;
 
     private final StateBackingStore<FileObject> store;
@@ -104,6 +99,8 @@ public class DefaultFileSystemMonitor implements FileSystemMonitor {
 
     private final Predicate<FileObjectStatus> cleanablePredicate;
 
+    private final TaskFileOrder taskFileOrder;
+
     /**
      * Creates a new {@link DefaultFileSystemMonitor} instance.
      *
@@ -118,16 +115,19 @@ public DefaultFileSystemMonitor(final Long allowTasksReconfigurationAfterTimeout
                                     final GenericFileCleanupPolicy cleanPolicy,
                                     final Predicate<FileObjectStatus> cleanablePredicate,
                                     final SourceOffsetPolicy offsetPolicy,
-                                    final StateBackingStore<FileObject> store) {
+                                    final StateBackingStore<FileObject> store,
+                                    final TaskFileOrder taskFileOrder) {
         Objects.requireNonNull(fsListening, "'fsListening' should not be null");
         Objects.requireNonNull(cleanPolicy, "'cleanPolicy' should not be null");
         Objects.requireNonNull(offsetPolicy, "'offsetPolicy' should not be null");
         Objects.requireNonNull(store, "'store' should not null");
         Objects.requireNonNull(cleanablePredicate, "'cleanablePredicate' should not null");
+        Objects.requireNonNull(taskFileOrder, "'taskFileOrder' should not null");
 
         this.fsListing = fsListening;
         this.allowTasksReconfigurationAfterTimeoutMs = allowTasksReconfigurationAfterTimeoutMs;
         this.cleanablePredicate = cleanablePredicate;
+        this.taskFileOrder = taskFileOrder;
 
         if (cleanPolicy instanceof FileCleanupPolicy) {
             this.cleaner = new DelegateBatchFileCleanupPolicy((FileCleanupPolicy) cleanPolicy);
@@ -420,12 +420,7 @@ public List<FileObjectMeta> listFilesToSchedule(final int maxFilesToSchedule) {
                     "or no object file was detected after starting the connector."
                 );
             }
-
-            // Sort all object files by the last-modified date before returning
-            return partitions
-                    .stream()
-                    .sorted(BY_LAST_MODIFIED)
-                    .collect(Collectors.toList());
+            return taskFileOrder.sort(partitions);
         } finally {
             scanned.clear();
             taskReconfigurationRequested.set(false);

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

@@ -54,6 +54,8 @@ public class DelegateTaskFileURIProvider implements TaskFileURIProvider {
 
     private StateBackingStoreAccess sharedStore;
 
+    private TaskFileOrder taskFileOrder;
+
     private StateSnapshot<FileObject> fileState;
 
     private boolean isFirstCall = true;
@@ -80,6 +82,7 @@ public void configure(final Map<String, ?> configs) {
         fileSystemListing = config.getFileSystemListing();
         sourceOffsetPolicy = config.getSourceOffsetPolicy();
         fileSystemListing.setFilter(new CompositeFileListFilter(config.getFileSystemListingFilter()));
+        taskFileOrder = config.getTaskFilerOrder();
     }
 
     /**
@@ -110,7 +113,9 @@ public List<URI> nextURIs() {
         ).values();
 
         isFirstCall = false;
-        return partitioner.partitionForTask(filtered, config.getTaskCount(), config.getTaskId());
+
+        List<FileObjectMeta> sorted = taskFileOrder.sort(filtered);
+        return partitioner.partitionForTask(sorted, config.getTaskCount(), config.getTaskId());
     }
 
     private void refreshState() {

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2022 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamthoughts.kafka.connect.filepulse.fs;
+
+import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Use for ordering files before assigning to task.
+ */
+public interface TaskFileOrder extends Function<Collection<FileObjectMeta>, List<FileObjectMeta>> {
+
+    static TaskFileOrder findBuiltInByName(final String name) {
+        return BuiltIn.valueOf(name).get();
+    }
+
+    List<FileObjectMeta> sort(final Collection<FileObjectMeta> objects);
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    default List<FileObjectMeta> apply(final Collection<FileObjectMeta> objects) {
+        return sort(objects);
+    }
+
+     enum BuiltIn {
+
+         /** Sort all object files by the last-modified date. */
+         LAST_MODIFIED(
+                 new AbstractTaskFileOrder(Comparator.comparingLong(FileObjectMeta::lastModified))
+         ),
+         /** Sort all object files by URI. */
+         URI(
+                 new AbstractTaskFileOrder(Comparator.comparing(FileObjectMeta::uri))
+         ),
+         /** Sort all object files by content-length. */
+         CONTENT_LENGTH(
+                 new AbstractTaskFileOrder(Comparator.comparingLong(FileObjectMeta::contentLength))
+         ),
+         /** Sort all object files by content-length. */
+         CONTENT_LENGTH_DESC(
+                 new AbstractTaskFileOrder(Comparator.comparingLong(FileObjectMeta::contentLength).reversed())
+         );
+
+         final TaskFileOrder order;
+
+         BuiltIn(TaskFileOrder order) {
+             this.order = order;
+         }
+
+         TaskFileOrder get() {
+             return order;
+         }
+
+    }
+
+    class AbstractTaskFileOrder implements TaskFileOrder {
+
+        private final Comparator<FileObjectMeta> comparator;
+
+        protected AbstractTaskFileOrder(Comparator<FileObjectMeta> comparator) {
+            this.comparator = comparator;
+        }
+
+        /**
+         * {@inheritDoc}
+         */
+        @Override
+        public List<FileObjectMeta> sort(final Collection<FileObjectMeta> objects) {
+            return objects
+                    .stream()
+                    .sorted(comparator)
+                    .collect(Collectors.toList());
+        }
+    }
+}

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

@@ -121,7 +121,8 @@ public void start(final Map<String, String> props) {
                     connectorConfig.getFsCleanupPolicy(),
                     connectorConfig.getFsCleanupPolicyPredicate(),
                     connectorConfig.getSourceOffsetPolicy(),
-                    sharedStore.get().getResource()
+                    sharedStore.get().getResource(),
+                    connectorConfig.getTaskFilerOrder()
             );
 
             monitor.setFileSystemListingEnabled(!connectorConfig.isFileListingTaskDelegationEnabled());

connect-file-pulse-plugin/src/test/java/io/streamthoughts/kafka/connect/filepulse/fs/DefaultFileSystemMonitorTest.java

@@ -255,7 +255,8 @@ private DefaultFileSystemMonitor newFileSystemMonitor(final MockFileCleaner clea
                 cleaner,
                 status -> List.of(FileObjectStatus.FAILED, FileObjectStatus.COMPLETED).contains(status),
                 OFFSET_MANAGER,
-                store
+                store,
+                TaskFileOrder.BuiltIn.LAST_MODIFIED.get()
         );
     }