feat(plugin): add new prop to configure if task should halt on error (#164)

Resolves: #164

Author: fhussonnois

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

@@ -50,7 +50,10 @@ public class CommonSourceConfig extends AbstractConfig {
     private static final String FS_SCAN_FILTERS_DOC           = "Filters classes which are used to apply list input files.";
 
     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.";
+    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_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 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";
@@ -61,10 +64,10 @@ public class CommonSourceConfig extends AbstractConfig {
     private static final String FILTER_DOC = "List of filters aliases to apply on each value (order is important).";
 
     public static final String TASKS_FILE_STATUS_STORAGE_CLASS_CONFIG = "tasks.file.status.storage.class";
-    public static final String TASKS_FILE_STATUS_STORAGE_CLASS_DOC = "The FileObjectStateBackingStore class to be used for storing status state of file objects.";
+    private static final String TASKS_FILE_STATUS_STORAGE_CLASS_DOC = "The FileObjectStateBackingStore class to be used for storing status state of file objects.";
 
     public static final String TASK_PARTITIONER_CLASS_CONFIG = "task.partitioner.class";
-    public static final String TASK_PARTITIONER_CLASS_DOC    = "The TaskPartitioner to be used for partitioning files to tasks";
+    private static final String TASK_PARTITIONER_CLASS_DOC    = "The TaskPartitioner to be used for partitioning files to tasks";
 
     /**
      * Creates a new {@link CommonSourceConfig} instance.
@@ -100,6 +103,17 @@ public static ConfigDef getConfigDev() {
                         ConfigDef.Width.NONE,
                         TASKS_FILE_READER_CLASS_CONFIG
                 )
+                .define(
+                        TASKS_HALT_ON_ERROR_CONFIG,
+                        ConfigDef.Type.BOOLEAN,
+                        false,
+                        ConfigDef.Importance.HIGH,
+                        TASKS_HALT_ON_ERROR_DOC,
+                        GROUP,
+                        groupCounter++,
+                        ConfigDef.Width.NONE,
+                        TASKS_HALT_ON_ERROR_CONFIG
+                )
                 .define(
                         OUTPUT_TOPIC_CONFIG,
                         ConfigDef.Type.STRING,
@@ -155,8 +169,7 @@ public static ConfigDef getConfigDev() {
                         TASK_PARTITIONER_CLASS_DOC
                 );
     }
-
-
+    
     public FileSystemListing<?> getFileSystemListing() {
         return getConfiguredInstance(FS_LISTING_CLASS_CONFIG, FileSystemListing.class);
     }
@@ -169,6 +182,10 @@ public TaskPartitioner getTaskPartitioner() {
         return this.getConfiguredInstance(TASK_PARTITIONER_CLASS_CONFIG, TaskPartitioner.class);
     }
 
+    public boolean isTaskHaltOnError() {
+        return this.getBoolean(TASKS_HALT_ON_ERROR_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/DefaultFileRecordsPollingConsumer.java

@@ -20,7 +20,6 @@
 
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.errors.ConnectFilePulseException;
-import io.streamthoughts.kafka.connect.filepulse.filter.FilterException;
 import io.streamthoughts.kafka.connect.filepulse.filter.RecordFilterPipeline;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputIterator;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputReader;
@@ -52,7 +51,7 @@ public class DefaultFileRecordsPollingConsumer implements FileRecordsPollingCons
     private final SourceTaskContext taskContext;
     private final AtomicBoolean closed = new AtomicBoolean(false);
 
-    private FileRecord<TypedStruct> latestPollRecord;
+    private FileRecord<TypedStruct> latestPolledRecord;
 
     private FileInputIterator<FileRecord<TypedStruct>> currentIterator;
 
@@ -134,11 +133,11 @@ void addAll(final List<URI> files) {
     public FileContext context() {
         if (currentIterator != null) {
             FileContext context = currentIterator.context();
-            if (latestPollRecord != null) {
+            if (latestPolledRecord != null) {
                 context = new FileContext(
                         context.key(),
                         context.metadata(),
-                        latestPollRecord.offset().toSourceOffset());
+                        latestPolledRecord.offset().toSourceOffset());
             }
             return context;
         }
@@ -162,39 +161,31 @@ public RecordsIterable<FileRecord<TypedStruct>> next() {
             throw new IllegalStateException("FileRecordsPollingConsumer is closed, no more element can be returned");
         }
 
-        if (!hasNext()) return RecordsIterable.empty();
+        if (!hasNext())
+            return RecordsIterable.empty();
 
-        // Quickly iterate to lookup for a valid iterator
-        do {
-            final DelegateFileInputIterator candidate = queue.peek();
-            if (candidate.isOpen()) {
-                currentIterator = getOrCloseIteratorIfNoMoreRecord(candidate);
-            } else {
-                currentIterator = openAndGetIteratorOrNullIfInvalid(taskContext, candidate);
-            }
-        } while (hasNext() && currentIterator == null);
+        currentIterator = findNextFileObjectIterator();
 
-        if (currentIterator == null) {
+        if (currentIterator == null)
             return RecordsIterable.empty();
-        }
-
-        final RecordsIterable<FileRecord<TypedStruct>> records = currentIterator.next();
 
         Exception exception = null;
         try {
+            // Read the next records from the current iterator
+            final RecordsIterable<FileRecord<TypedStruct>> records = currentIterator.next();
+
+            // Apply the filter-chain in the returned records
             final RecordsIterable<FileRecord<TypedStruct>> filtered = pipeline.apply(
                     records,
                     currentIterator.hasNext()
             );
+            // May update the last polled records.
             if (!filtered.isEmpty()) {
-                latestPollRecord = filtered.last();
+                latestPolledRecord = filtered.last();
             }
-            return filtered;
-        } catch (final FilterException e) {
-            exception = e;
-            // ignore the error - and skip the current file.
-            return RecordsIterable.empty();
 
+            // Return record to the connect SourceTask
+            return filtered;
         } catch (final ConnectFilePulseException e) {
             exception = e;
             throw e;
@@ -210,6 +201,40 @@ public RecordsIterable<FileRecord<TypedStruct>> next() {
         }
     }
 
+    private FileInputIterator<FileRecord<TypedStruct>> findNextFileObjectIterator() {
+
+        if (queue.isEmpty()) return null;
+
+        FileInputIterator<FileRecord<TypedStruct>> ret;
+        // Quickly iterate to lookup for a valid iterator
+        do {
+            final DelegateFileInputIterator candidate = queue.peek();
+            if (candidate.isOpen()) {
+                ret = getOrCloseIteratorIfNoMoreRecord(candidate);
+            } else {
+                try {
+                    ret = openAndGetIteratorOrNullIfInvalid(candidate);
+                    if (ret == null) {
+                        var objectMeta = new GenericFileObjectMeta(candidate.getObjectURI());
+                        // Remove the current iterator and continue
+                        deleteFileQueueAndInvokeListener(new FileContext(candidate.key(), objectMeta), null);
+                    }
+                } catch (Exception e) {
+                    LOG.error(
+                        "Failed to open and initialize new iterator for object: {}.",
+                        candidate.getObjectURI()
+                    );
+                    final FileObjectMeta objectMeta = new GenericFileObjectMeta(candidate.getObjectURI());
+                    deleteFileQueueAndInvokeListener(new FileContext(candidate.key(), objectMeta), e);
+                    // Rethrow the exception.
+                    throw e;
+                }
+            }
+        } while (hasNext() && ret == null);
+
+        return ret;
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -259,19 +284,16 @@ public void setStateListener(final StateListener listener) {
      * This method will return {@code null} if the iterator point
      * to either an invalid file or to an already been completed file.
      *
-     * @param context  the connect source task context
      * @param iterator the source file iterator
      * @return a new {@link FileInputIterator} instance or {@code null} if the iterator is invalid.
      */
     private FileInputIterator<FileRecord<TypedStruct>> openAndGetIteratorOrNullIfInvalid(
-            final SourceTaskContext context,
             final DelegateFileInputIterator iterator
     ) {
         // Re-check if the file is still valid.
         if (!iterator.isValid()) {
-            LOG.error("File does not exist or is not readable, skip entry and continue '{}'", iterator.getObjectURI());
-            var objectMeta = new GenericFileObjectMeta(iterator.getObjectURI());
-            deleteFileQueueAndInvokeListener(new FileContext(iterator.key(), objectMeta), null);
+            LOG.warn("File does not exist or is not readable. Skip entry and continue '{}'", iterator.getObjectURI());
+            // Return NULL so that the calling method can properly close the iterator.
             return null;
         }
 
@@ -282,58 +304,42 @@ private FileInputIterator<FileRecord<TypedStruct>> openAndGetIteratorOrNullIfInv
 
             if (!ignoreCommittedOffsets) {
                 committedOffset = offsetPolicy
-                        .getOffsetFor(context, metadata)
+                        .getOffsetFor(taskContext, metadata)
                         .orElse(FileObjectOffset.empty());
             } else {
                 committedOffset = FileObjectOffset.empty();
             }
         } catch (final Exception e) {
-            if (metadata == null) {
-                LOG.error(
-                        "Failed to load metadata for object file {}. Error: {}",
-                        iterator.getObjectURI(),
-                        e.getMessage()
-                );
-                metadata = new GenericFileObjectMeta(iterator.getObjectURI());
-                deleteFileQueueAndInvokeListener(new FileContext(iterator.key(), metadata), e);
-                return null;
-            } else {
+            if (metadata != null) {
                 LOG.warn(
-                        "Failed to load committed offset for object file {}." +
-                        " Previous offset will be ignored. Error: {}",
+                        "Failed to load committed offset for object file {}. " +
+                        "Previous offset will be ignored. Error: {}",
                         iterator.getObjectURI(),
                         e.getMessage()
                 );
                 committedOffset = FileObjectOffset.empty();
+            } else {
+                // Rethrow the exception if metadata cannot be loaded.
+                // The exception handling logic will be delegated to the calling method.
+                throw e;
+
             }
         }
 
         // Quickly check if we can considered this file completed based on the content-length.
         boolean isAlreadyCompleted = committedOffset.position() >= metadata.contentLength();
         if (!ignoreCommittedOffsets && isAlreadyCompleted) {
             LOG.warn(
-                    "Detected source file already completed, skip entry and continue '{}'",
-                    iterator.getObjectURI());
-            deleteFileQueueAndInvokeListener(
-                    new FileContext(iterator.key(), metadata, committedOffset),
-                    null
+                "Detected source file already completed. Skip entry and continue '{}'",
+                iterator.getObjectURI()
             );
+            // Return NULL so that the calling method can properly close the iterator.
             return null;
         }
 
-        try {
-            iterator.open();
-            iterator.seekTo(committedOffset);
-            pipeline.init(iterator.context());
-        } catch (final Exception e) {
-            LOG.error(
-                    "Failed to initialized a new iterator for object file '{}'.",
-                    iterator.getObjectURI(),
-                    e
-            );
-            deleteFileQueueAndInvokeListener(new FileContext(iterator.key(), metadata), e);
-            return null;
-        }
+        iterator.open();
+        iterator.seekTo(committedOffset);
+        pipeline.init(iterator.context());
         if (hasListener()) {
             listener.onStart(iterator.context());
         }
@@ -365,14 +371,14 @@ private void closeIterator(final FileInputIterator<FileRecord<TypedStruct>> iter
         }
     }
 
-    private void deleteFileQueueAndInvokeListener(final FileContext taskContext,
+    private void deleteFileQueueAndInvokeListener(final FileContext fileContext,
                                                   final Throwable exception) {
         queue.remove();
         if (hasListener()) {
             if (exception != null) {
-                listener.onFailure(taskContext, exception);
+                listener.onFailure(fileContext, exception);
             } else {
-                listener.onCompleted(taskContext);
+                listener.onCompleted(fileContext);
             }
         }
     }

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

@@ -174,22 +174,29 @@ public List<SourceRecord> poll() throws InterruptedException {
                     }
                 } else {
 
-                    final RecordsIterable<FileRecord<TypedStruct>> records = consumer.next();
-
-                    if (!records.isEmpty()) {
-                        final FileContext context = consumer.context();
-                        LOG.debug("Returning {} records for {}", records.size(), context.metadata());
-                        results = records.stream()
-                                .map(r -> buildSourceRecord(context, r))
-                                .collect(Collectors.toList());
-
-                    // Check if the SourceTask is still running to
-                    // return immediately instead of waiting
-                    } else if (running.get() &&
-                               consumer.hasNext() &&
-                               consecutiveWaits.checkAndDecrement()) {
-                        busyWait();
-                        continue;
+                    try {
+                        final RecordsIterable<FileRecord<TypedStruct>> records = consumer.next();
+                        if (!records.isEmpty()) {
+                            final FileContext context = consumer.context();
+                            LOG.debug("Returning {} records for {}", records.size(), context.metadata());
+                            results = records.stream()
+                                    .map(r -> buildSourceRecord(context, r))
+                                    .collect(Collectors.toList());
+
+                            // Check if the SourceTask is still running to
+                            // return immediately instead of waiting
+                        } else if (running.get() &&
+                                consumer.hasNext() &&
+                                consecutiveWaits.checkAndDecrement()) {
+                            busyWait();
+                            continue;
+                        }
+                    } catch (ConnectFilePulseException e) {
+                        if (taskConfig.isTaskHaltOnError()) {
+                            throw e;
+                        } else {
+                            LOG.error("Caught unexpected error while processing file. Ignore and continue", e);
+                        }
                     }
                 }
 

connect-file-pulse-plugin/src/test/java/io/streamthoughts/kafka/connect/filepulse/config/SourceTaskConfigTest.java

@@ -41,7 +41,7 @@ public void should_configure_filter_given_on_failure() {
         final SourceTaskConfig config = new SourceTaskConfig(new HashMap<>() {{
             put(CommonSourceConfig.OUTPUT_TOPIC_CONFIG, "Test");
             put(CommonSourceConfig.FS_LISTING_CLASS_CONFIG, MockFileSystemListing.class.getName());
-            put(CommonSourceConfig.TASKS_FILE_READER_CLASS_CONFIG, LocalRowFileInputReader.class.getName());
+            put(CommonSourceConfig.TASKS_HALT_ON_ERROR, LocalRowFileInputReader.class.getName());
             put(DefaultTaskFileURIProvider.Config.FILE_OBJECT_URIS_CONFIG, "/tmp");
             put(CommonSourceConfig.FILTER_CONFIG, "Test");
             put(CommonSourceConfig.FILTER_CONFIG + ".Test.type", AppendFilter.class.getName());