fix(plugin): files should be cleanup when an error occurred while converting data to connect record

Author: fhussonnois

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

@@ -182,7 +182,7 @@ public RecordsIterable<FileRecord<TypedStruct>> next() {
                 latestPolledRecord = filtered.last();
             }
 
-            // Return record to the connect SourceTask
+            // Return record to SourceTask
             return filtered;
         } catch (final ConnectFilePulseException e) {
             exception = e;
@@ -325,7 +325,7 @@ private FileInputIterator<FileRecord<TypedStruct>> openAndGetIteratorOrNullIfCom
             }
         }
 
-        // Quickly check if we can considered this file completed based on the content-length.
+        // Quickly check if we can consider this file completed based on the content-length.
         boolean isAlreadyCompleted = committedOffset.position() >= metadata.contentLength();
         if (!ignoreCommittedOffsets && isAlreadyCompleted) {
             LOG.warn(
@@ -358,6 +358,10 @@ private FileInputIterator<FileRecord<TypedStruct>> getOrCloseIteratorIfNoMoreRec
         return null;
     }
 
+    public void closeCurrentIterator(final Exception cause) {
+        closeIterator(currentIterator, cause);
+    }
+
     private void closeIterator(final FileInputIterator<FileRecord<TypedStruct>> iterator,
                                final Exception cause) {
         final FileContext context = iterator.context();

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

@@ -295,10 +295,15 @@ private SourceRecord buildSourceRecord(final FileContext context,
             return result;
 
         } catch (final Throwable t) {
-            throw new ConnectFilePulseException(
-                    "Failed to convert data into connect record: '" + context.metadata().uri() + "'",
+            var exception = new ConnectFilePulseException(String.format(
+                    "Failed to convert data into Kafka Connect record at offset %s from object-file: %s'",
+                    context.offset(),
+                    context.metadata()),
                     t
             );
+            // Close internal iterator for the current object-file so that it will be marked as failed
+            consumer.closeCurrentIterator(exception);
+            throw exception;
         }
     }