subtask(all): refactor FilePulse API to support remote storages (#100)

This commit contains a large refactoring of the Connect FilePulse API to remove
the used of java class File for referencing an input file. This is
prerequisite for Support streaming files from a remote storage Blob, S3, etc.

Breaking changes:
- add new methods to FileInputReader: readMetadata, isAccessible
- update and rename interface FSDirectoryWalker to FileSystemListing
- update interface FileListFilter to not use the File class
- refactor class OffsetManager to SourceOffsetPolicy
- add new class DefaultOffsetPolicy

- rename SourceFile to FileObject
- rename SourceOffset to FileObjectOffset
- rename SourceStatus to FileObjectStatus
- rename SourceMetadara to FileObjectMeta

- add new top-level module filepulse-filesystems
- add new module filepulse-local-fs
- relocate readers for local filesystems

Change configuration properties:
- fs.scanner.class ->  fs.listing.class
- fs.scan.filers -> fs.listing.filters
- fs.scan.directory.path -> fs.listing.directory.path
- fs.recursive.scan.enable -> fs.listing.recursive.enabled

Resolves: #100

Author: fhussonnois

checkstyle/suppressions.xml

@@ -53,8 +53,9 @@
     <suppress checks="LineLength" files="ConnectorConfig.java"/>
     <suppress checks="LineLength" files="TaskConfig.java"/>
     <suppress checks="LineLength" files="CommonConfig.java"/>
-    <suppress checks="LineLength" files="LocalFileSystemScanner.java"/>
+    <suppress checks="LineLength" files="DefaultFileSystemScanner.java"/>
     <suppress checks="LineLength" files="ComposeOffsetStrategy.java"/>
+    <suppress checks="LineLength" files="DefaultOffsetPolicyConfig.java"/>
     <!-- Those classes are copy from kafka-connect api-->
     <suppress checks="LineLength" files="KafkaBasedLog.java"/>
     <suppress checks="NPathComplexity" files="KafkaBasedLog.java"/>

config/connect-file-pulse-example-override-topic-and-key.json

@@ -12,14 +12,14 @@
     "filters.ParseDelimitedRow.trimColumn": "true",
     "filters.ParseDelimitedRow.type": "io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter",
     "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
-    "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
+    "fs.listing.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.LocalFSDirectoryListing",
     "fs.scan.directory.path": "/tmp/kafka-connect/examples/",
     "fs.scan.interval.ms": "10000",
     "internal.kafka.reporter.bootstrap.servers": "broker:29092",
     "internal.kafka.reporter.topic": "connect-file-pulse-status",
     "offset.strategy": "name+hash",
     "skip.headers": "1",
-    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
+    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.reader.RowFileInputReader",
     "topic": "connect-file-pulse-quickstart-csv",
     "tasks.max": 1
   },

config/connect-file-pulse-quickstart-avro.json

@@ -2,14 +2,14 @@
   "config": {
     "connector.class": "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
     "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
-    "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
+    "fs.listing.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.LocalFSDirectoryListing",
     "fs.scan.directory.path": "/tmp/kafka-connect/examples/",
     "fs.scan.interval.ms": "10000",
     "internal.kafka.reporter.bootstrap.servers": "broker:29092",
     "internal.kafka.reporter.topic": "connect-file-pulse-status",
     "offset.strategy": "name",
     "read.max.wait.ms": "5000",
-    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.AvroFileInputReader",
+    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.reader.AvroFileInputReader",
     "topic": "connect-file-pulse-quickstart-avro",
     "tasks.max": 1
   },

config/connect-file-pulse-quickstart-csv.json

@@ -9,14 +9,14 @@
     "filters.ParseDelimitedRow.trimColumn": "true",
     "filters.ParseDelimitedRow.type": "io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter",
     "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
-    "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
+    "fs.listing.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.LocalFSDirectoryListing",
     "fs.scan.directory.path": "/tmp/kafka-connect/examples/",
     "fs.scan.interval.ms": "10000",
     "internal.kafka.reporter.bootstrap.servers": "broker:29092",
     "internal.kafka.reporter.topic": "connect-file-pulse-status",
     "offset.strategy": "name+hash",
     "skip.headers": "1",
-    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
+    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.reader.RowFileInputReader",
     "topic": "connect-file-pulse-quickstart-csv",
     "tasks.max": 1
   },

config/connect-file-pulse-quickstart-log4j.json

@@ -11,14 +11,14 @@
     "filters.ParseLog4jLog.type": "io.streamthoughts.kafka.connect.filepulse.filter.GrokFilter",
     "filters.ParseLog4jLog.ignoreFailure": "true",
     "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
-    "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
+    "fs.listing.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.LocalFSDirectoryListing",
     "fs.scan.directory.path": "/var/log/kafka/",
     "fs.scan.interval.ms": "10000",
     "internal.kafka.reporter.bootstrap.servers": "broker:29092",
     "internal.kafka.reporter.topic": "connect-file-pulse-status",
     "offset.strategy": "name",
     "read.max.wait.ms": "5000",
-    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
+    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.fs.local.reader.RowFileInputReader",
     "topic": "connect-file-pulse-quickstart-log4j",
     "tasks.max": 1
   },

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/clean/BatchFileCleanupPolicy.java

@@ -18,13 +18,14 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.clean;
 
-import io.streamthoughts.kafka.connect.filepulse.source.SourceFile;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObject;
 
 import java.util.List;
 
 /**
  * Policy for cleaning a batch of completed source files.
  */
-public interface BatchFileCleanupPolicy extends GenericFileCleanupPolicy<List<SourceFile>,FileCleanupPolicyResultSet> {
+public interface BatchFileCleanupPolicy
+        extends GenericFileCleanupPolicy<List<FileObject>,FileCleanupPolicyResultSet> {
 
 }

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/clean/DelegateBatchFileCleanupPolicy.java

@@ -18,7 +18,7 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.clean;
 
-import io.streamthoughts.kafka.connect.filepulse.source.SourceFile;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObject;
 
 import java.util.List;
 import java.util.Map;
@@ -50,9 +50,9 @@ public void configure(final Map<String, ?> configs) {
      * {@inheritDoc}
      */
     @Override
-    public FileCleanupPolicyResultSet apply(final List<SourceFile> sources) {
+    public FileCleanupPolicyResultSet apply(final List<FileObject> sources) {
         FileCleanupPolicyResultSet rs = new FileCleanupPolicyResultSet();
-        for (SourceFile source : sources) {
+        for (FileObject source : sources) {
             if (delegate.apply(source)) {
                 rs.add(source, FileCleanupPolicyResult.SUCCEED);
             } else {

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/clean/FileCleanupPolicy.java

@@ -18,23 +18,23 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.clean;
 
-import io.streamthoughts.kafka.connect.filepulse.source.SourceFile;
-import io.streamthoughts.kafka.connect.filepulse.source.SourceStatus;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObject;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObjectStatus;
 
 /**
  * Policy for cleaning individual completed source files.
  */
-public interface FileCleanupPolicy extends GenericFileCleanupPolicy<SourceFile, Boolean> {
+public interface FileCleanupPolicy extends GenericFileCleanupPolicy<FileObject, Boolean> {
 
     /**
      * {@inheritDoc}
      */
     @Override
-    default Boolean apply(final SourceFile source) {
-        return source.status().equals(SourceStatus.COMPLETED) ? onSuccess(source) : onFailure(source);
+    default Boolean apply(final FileObject source) {
+        return source.status().equals(FileObjectStatus.COMPLETED) ? onSuccess(source) : onFailure(source);
     }
 
-    boolean onSuccess(final SourceFile source);
+    boolean onSuccess(final FileObject source);
 
-    boolean onFailure(final SourceFile source);
+    boolean onFailure(final FileObject source);
 }

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/clean/FileCleanupPolicyResultSet.java

@@ -19,7 +19,7 @@
 package io.streamthoughts.kafka.connect.filepulse.clean;
 
 import io.streamthoughts.kafka.connect.filepulse.internal.KeyValuePair;
-import io.streamthoughts.kafka.connect.filepulse.source.SourceFile;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObject;
 
 import java.util.Collection;
 import java.util.LinkedList;
@@ -28,7 +28,7 @@
 
 public class FileCleanupPolicyResultSet {
 
-    private final Collection<KeyValuePair<SourceFile, FileCleanupPolicyResult>> resultSet;
+    private final Collection<KeyValuePair<FileObject, FileCleanupPolicyResult>> resultSet;
 
     /**
      * Creates a new {@link FileCleanupPolicyResultSet} instance.
@@ -38,15 +38,15 @@ public FileCleanupPolicyResultSet() {
     }
 
 
-    public void add(final SourceFile source, FileCleanupPolicyResult result) {
+    public void add(final FileObject source, FileCleanupPolicyResult result) {
         Objects.requireNonNull(source, "source cannot be null");
         Objects.requireNonNull(result, "result cannot be null");
         resultSet.add(KeyValuePair.of(source, result));
     }
 
-    public void forEach(final BiConsumer<? super SourceFile, FileCleanupPolicyResult> action) {
+    public void forEach(final BiConsumer<? super FileObject, FileCleanupPolicyResult> action) {
         Objects.requireNonNull(action);
-        for (KeyValuePair<SourceFile, FileCleanupPolicyResult> kv : resultSet) {
+        for (KeyValuePair<FileObject, FileCleanupPolicyResult> kv : resultSet) {
             action.accept(kv.key, kv.value);
         }
     }

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/DefaultRecordFilterPipeline.java

@@ -23,7 +23,7 @@
 import io.streamthoughts.kafka.connect.filepulse.reader.RecordsIterable;
 import io.streamthoughts.kafka.connect.filepulse.source.FileContext;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
-import io.streamthoughts.kafka.connect.filepulse.source.SourceMetadata;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
 import io.streamthoughts.kafka.connect.filepulse.source.TypedFileRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -102,7 +102,7 @@ public RecordsIterable<FileRecord<TypedStruct>> apply(final RecordsIterable<File
     }
 
     private FilterContext getContextFor(final FileRecord<TypedStruct> record,
-                                        final SourceMetadata metadata) {
+                                        final FileObjectMeta metadata) {
         return FilterContextBuilder
             .newBuilder()
             .withMetadata(metadata)