V2.0 (jcustenborder#106)

* Start of the 2.0 branch.

* Added support for sorting the order of files processed. Added support for multiple tasks. Refactored file selection logic out to predicates. Fixes jcustenborder#98. Fixes jcustenborder#97.

* Clean up policy to move files to subdirectories by date in finished directory (jcustenborder#103)

* Moving finished files to date folders

* Add test

* Removed duplicate call to listFiles.

* Moving the metadata to headers. Metadata currently was not used by the individual connectors. Fixed jcustenborder#94. Fixed jcustenborder#95.

* Refactored to remove usage of Path since we have that functionality in File. Lower logging of directory to trace.

* Refactored InputFile to own everything for the input file. Added metadata class that is responsible for the file metadata based on cached values from InputFile. Updated test cases to have values for headers, ignoring some of the headers that will by dynamic.

* Fixed java doc warnings.

* Added finishedPathRequired method to config. This will be used to determine if the tasks should check for a finished path. Moved ownership of the InputStream to be completely within InputFile.

* Fixed checkstyle.

* Major refactor to add support for using a BufferedInputStream. Renamed abstract classes to be consistent. Connectors can optionally support using a BufferedInputStream which should reduce the round trips to the underlying filesystem. `file.buffer.size.bytes` can be used to configure the buffer. Fixes jcustenborder#105.

* Corrected Metadata to include offset within the file. jcustenborder#95.

* Corrected to use proper config() method.

* Added missing file.offset header to test cases. jcustenborder#95.

* Removed duplicate check.

* Added MB per second calculations.

* Bump the backoff time to 500 ms if no files are returned.

* Lower the logging level when searching for files.

* Correct Connect Packaging plugin configuration (jcustenborder#101)

Correct Connect Packaging plugin configuration so that the project's packaging configuration properties properly override those inherited from the parent POM.

* Change manifest file's link to documentation (jcustenborder#102)

* Missed configuring tasks for SchemaLessJson. jcustenborder#97.

* Fixed Checkstyle.

* Refactored to allow metadata to be placed in a header, field, or dropped.

* Modified to not require a key for the messages. Bumped version for extended-log-format. Fixes jcustenborder#117

* CC-5455: Update support summary for LA connector (jcustenborder#109)

Signed-off-by: Arjun Satish <arjun@confluent.io>

* Refactored to allow metadata to be placed in a header, field, or dropped.

* Documentation for `metadata.field` was not correct.

* Corrected extended-log-format version.

* Update docs for task.partitioner.

* Removed schema generation dependency.

* Updated examples.

Co-authored-by: Jon Mitchell <44587662+jmitchellwf@users.noreply.github.com>
Co-authored-by: Randall Hauch <rhauch@gmail.com>
Co-authored-by: Arjun Satish <wicknicks@users.noreply.github.com>

Author: 4 people

.gitignore

@@ -2,3 +2,6 @@ target
 *.iml
 .okhttpcache
 ELFTesting.properties
+.checkstyle
+.factorypath
+.idea/

config/CSVExample.json

@@ -0,0 +1,17 @@
+{
+  "name": "CsvSpoolDir",
+  "config": {
+    "tasks.max": "1",
+    "connector.class": "com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector",
+    "input.file.pattern": "^.*\\.csv$",
+    "halt.on.error": "false",
+    "topic": "testing"
+    "csv.first.row.as.header": "true",
+    "csv.null.field.indicator": "EMPTY_SEPARATORS",
+    "input.path": "/tmp/spooldir/input",
+    "finished.path": "/tmp/spooldir/finished",
+    "error.path": "/tmp/spooldir/error",
+    "key.schema": "{\"name\":\"com.example.users.UserKey\",\"type\":\"STRUCT\",\"isOptional\":false,\"fieldSchemas\":{\"id\":{\"type\":\"INT64\",\"isOptional\":false}}}",
+    "value.schema": "{\"name\":\"com.example.users.User\",\"type\":\"STRUCT\",\"isOptional\":false,\"fieldSchemas\":{\"id\":{\"type\":\"INT64\",\"isOptional\":false},\"first_name\":{\"type\":\"STRING\",\"isOptional\":true},\"last_name\":{\"type\":\"STRING\",\"isOptional\":true},\"email\":{\"type\":\"STRING\",\"isOptional\":true},\"gender\":{\"type\":\"STRING\",\"isOptional\":true},\"ip_address\":{\"type\":\"STRING\",\"isOptional\":true},\"last_login\":{\"name\":\"org.apache.kafka.connect.data.Timestamp\",\"type\":\"INT64\",\"version\":1,\"isOptional\":true},\"account_balance\":{\"name\":\"org.apache.kafka.connect.data.Decimal\",\"type\":\"BYTES\",\"version\":1,\"parameters\":{\"scale\":\"2\"},\"isOptional\":true},\"country\":{\"type\":\"STRING\",\"isOptional\":true},\"favorite_color\":{\"type\":\"STRING\",\"isOptional\":true}}}"
+  }
+}

config/CSVExample.properties

@@ -18,9 +18,17 @@ name=CsvSpoolDir
 tasks.max=1
 connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector
 input.file.pattern=^.*\.csv$
-finished.path=/tmp/spooldir/finished
+
 halt.on.error=false
 topic=testing
 key.schema={"name":"com.example.users.UserKey","type":"STRUCT","isOptional":false,"fieldSchemas":{"id":{"type":"INT64","isOptional":false}}}
 value.schema={"name":"com.example.users.User","type":"STRUCT","isOptional":false,"fieldSchemas":{"id":{"type":"INT64","isOptional":false},"first_name":{"type":"STRING","isOptional":true},"last_name":{"type":"STRING","isOptional":true},"email":{"type":"STRING","isOptional":true},"gender":{"type":"STRING","isOptional":true},"ip_address":{"type":"STRING","isOptional":true},"last_login":{"name":"org.apache.kafka.connect.data.Timestamp","type":"INT64","version":1,"isOptional":true},"account_balance":{"name":"org.apache.kafka.connect.data.Decimal","type":"BYTES","version":1,"parameters":{"scale":"2"},"isOptional":true},"country":{"type":"STRING","isOptional":true},"favorite_color":{"type":"STRING","isOptional":true}}}
 csv.first.row.as.header=true
+csv.null.field.indicator=EMPTY_SEPARATORS
+
+input.path=/tmp/spooldir/input
+finished.path=/tmp/spooldir/finished
+error.path=/tmp/spooldir/error
+batch.size = 5000
+cleanup.policy = DELETE
+file.buffer.size.bytes = 1048576

config/connect-avro-docker.properties

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
+group.id=foo
 bootstrap.servers=kafka:9092
 key.converter=io.confluent.connect.avro.AvroConverter
 key.converter.schema.registry.url=http://schema-registry:8081
@@ -24,4 +24,11 @@ internal.value.converter=org.apache.kafka.connect.json.JsonConverter
 internal.key.converter.schemas.enable=false
 internal.value.converter.schemas.enable=false
 offset.storage.file.filename=/tmp/connect.offsets
-plugin.path=target/kafka-connect-target/usr/share/kafka-connect
+plugin.path=target/kafka-connect-target/usr/share/kafka-connect
+
+config.storage.replication.factor=1
+config.storage.topic=connect_config
+offset.storage.replication.factor=1
+offset.storage.topic=connect_offset
+status.storage.replication.factor=1
+status.storage.topic=connect_status

pom.xml

@@ -61,6 +61,11 @@
         <url>https://github.com/jcustenborder/kafka-connect-spooldir/issues</url>
     </issueManagement>
     <dependencies>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
         <dependency>
             <groupId>net.sourceforge.argparse4j</groupId>
             <artifactId>argparse4j</artifactId>
@@ -72,6 +77,12 @@
             <artifactId>opencsv</artifactId>
             <version>4.5</version>
         </dependency>
+        <dependency>
+            <groupId>io.confluent</groupId>
+            <artifactId>kafka-connect-avro-converter</artifactId>
+            <version>5.2.1</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-compress</artifactId>
@@ -80,7 +91,7 @@
         <dependency>
             <groupId>com.github.jcustenborder.parsers</groupId>
             <artifactId>extended-log-format</artifactId>
-            <version>[0.0.1.2, 0.0.1.1000)</version>
+            <version>[0.0.2.12, 0.0.2.1000)</version>
         </dependency>
     </dependencies>
     <build>

src/main/java/com/github/jcustenborder/kafka/connect/spooldir/AbstractCleanUpPolicy.java

@@ -15,22 +15,23 @@
  */
 package com.github.jcustenborder.kafka.connect.spooldir;
 
-import com.google.common.io.Files;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
+import java.text.SimpleDateFormat;
 
 abstract class AbstractCleanUpPolicy implements Closeable {
   private static final Logger log = LoggerFactory.getLogger(AbstractCleanUpPolicy.class);
-  protected final File inputFile;
+  private static SimpleDateFormat dateFormatter = new SimpleDateFormat("yyyy-MM-dd");
+  protected final InputFile inputFile;
   protected final File errorPath;
   protected final File finishedPath;
 
 
-  protected AbstractCleanUpPolicy(File inputFile, File errorPath, File finishedPath) {
+  protected AbstractCleanUpPolicy(InputFile inputFile, File errorPath, File finishedPath) {
     this.inputFile = inputFile;
     this.errorPath = errorPath;
     this.finishedPath = finishedPath;
@@ -41,13 +42,16 @@ public static AbstractCleanUpPolicy create(AbstractSourceConnectorConfig config,
     final AbstractCleanUpPolicy result;
     switch (config.cleanupPolicy) {
       case MOVE:
-        result = new Move(inputFile.inputFile, config.errorPath, config.finishedPath);
+        result = new Move(inputFile, config.errorPath, config.finishedPath);
+        break;
+      case MOVEBYDATE:
+        result = new MoveByDate(inputFile, config.errorPath, config.finishedPath);
         break;
       case DELETE:
-        result = new Delete(inputFile.inputFile, config.errorPath, config.finishedPath);
+        result = new Delete(inputFile, config.errorPath, config.finishedPath);
         break;
       case NONE:
-        result = new None(inputFile.inputFile, config.errorPath, config.finishedPath);
+        result = new None(inputFile, config.errorPath, config.finishedPath);
         break;
       default:
         throw new UnsupportedOperationException(
@@ -58,23 +62,22 @@ public static AbstractCleanUpPolicy create(AbstractSourceConnectorConfig config,
     return result;
   }
 
-  protected void removeFile(File file) {
-    log.info("Removing {}", file);
-    if (!file.delete()) {
-      log.warn("Could not delete {}", file);
-    }
-  }
 
-  protected void moveToDirectory(File outputDirectory) {
-    File outputFile = new File(outputDirectory, this.inputFile.getName());
-    try {
-      if (this.inputFile.exists()) {
-        log.info("Moving {} to {}", this.inputFile, outputFile);
-        Files.move(this.inputFile, outputFile);
-      }
-    } catch (IOException e) {
-      log.error("Exception thrown while trying to move {} to {}", this.inputFile, outputFile, e);
+
+
+  protected boolean createDirectory(File directory) {
+    if (directory.exists()) {
+      return true;
+    }
+    if (!directory.mkdir()) {
+      log.error("Cannot make directory - " + directory.getAbsolutePath());
+      return false;
     }
+    if (!directory.setWritable(true)) {
+      log.error("Cannot make directory writable - " + directory.getAbsolutePath());
+      return false;
+    }
+    return true;
   }
 
   @Override
@@ -91,7 +94,7 @@ public void error() {
         this.inputFile,
         this.errorPath
     );
-    moveToDirectory(this.errorPath);
+    this.inputFile.moveToDirectory(this.errorPath);
   }
 
   /**
@@ -100,29 +103,48 @@ public void error() {
   public abstract void success() throws IOException;
 
   static class Move extends AbstractCleanUpPolicy {
-    protected Move(File inputFile, File errorPath, File finishedPath) {
+    protected Move(InputFile inputFile, File errorPath, File finishedPath) {
       super(inputFile, errorPath, finishedPath);
     }
 
     @Override
     public void success() throws IOException {
-      moveToDirectory(this.finishedPath);
+      this.inputFile.moveToDirectory(this.finishedPath);
+    }
+  }
+
+  static class MoveByDate extends AbstractCleanUpPolicy {
+    protected MoveByDate(InputFile inputFile, File errorPath, File finishedPath) {
+      super(inputFile, errorPath, finishedPath);
+    }
+
+    @Override
+    public void success() throws IOException {
+      // Setup directory named as the file created date
+      File subDirectory = new File(this.finishedPath, dateFormatter.format(this.inputFile.lastModified()));
+      log.trace("Finished path: {}", subDirectory);
+
+      if (createDirectory(subDirectory)) {
+        this.inputFile.moveToDirectory(subDirectory);
+      } else {
+        this.inputFile.moveToDirectory(this.finishedPath);
+      }
     }
   }
 
   static class Delete extends AbstractCleanUpPolicy {
-    protected Delete(File inputFile, File errorPath, File finishedPath) {
+    protected Delete(InputFile inputFile, File errorPath, File finishedPath) {
       super(inputFile, errorPath, finishedPath);
     }
 
     @Override
     public void success() throws IOException {
-      removeFile(this.inputFile);
+      this.inputFile.delete();
     }
   }
 
   static class None extends AbstractCleanUpPolicy {
-    protected None(File inputFile, File errorPath, File finishedPath) {
+    protected None(InputFile inputFile, File errorPath, File finishedPath) {
       super(inputFile, errorPath, finishedPath);
     }
 

src/main/java/com/github/jcustenborder/kafka/connect/spooldir/SchemaGenerator.java renamed to src/main/java/com/github/jcustenborder/kafka/connect/spooldir/AbstractSchemaGenerator.java

@@ -40,10 +40,10 @@
 import java.util.Map;
 import java.util.Properties;
 
-public abstract class SchemaGenerator<CONFIG extends SpoolDirSourceConnectorConfig> {
+public abstract class AbstractSchemaGenerator<CONFIG extends AbstractSpoolDirSourceConnectorConfig> {
   static final String DUMMY_SCHEMA;
   static final Map<String, Object> DEFAULTS;
-  private static final Logger log = LoggerFactory.getLogger(SchemaGenerator.class);
+  private static final Logger log = LoggerFactory.getLogger(AbstractSchemaGenerator.class);
 
   static {
     String dummySchema;
@@ -62,17 +62,17 @@ public abstract class SchemaGenerator<CONFIG extends SpoolDirSourceConnectorConf
     defaultSettings.put(AbstractSourceConnectorConfig.INPUT_PATH_CONFIG, "/tmp/input");
     defaultSettings.put(AbstractSourceConnectorConfig.FINISHED_PATH_CONFIG, "/tmp/finish");
     defaultSettings.put(AbstractSourceConnectorConfig.ERROR_PATH_CONFIG, "/tmp/error");
-    defaultSettings.put(SpoolDirSourceConnectorConfig.VALUE_SCHEMA_CONF, DUMMY_SCHEMA);
-    defaultSettings.put(SpoolDirSourceConnectorConfig.KEY_SCHEMA_CONF, DUMMY_SCHEMA);
+    defaultSettings.put(AbstractSpoolDirSourceConnectorConfig.VALUE_SCHEMA_CONF, DUMMY_SCHEMA);
+    defaultSettings.put(AbstractSpoolDirSourceConnectorConfig.KEY_SCHEMA_CONF, DUMMY_SCHEMA);
     defaultSettings.put(AbstractSourceConnectorConfig.TOPIC_CONF, "dummy");
-    defaultSettings.put(SpoolDirSourceConnectorConfig.SCHEMA_GENERATION_ENABLED_CONF, "true");
+    defaultSettings.put(AbstractSpoolDirSourceConnectorConfig.SCHEMA_GENERATION_ENABLED_CONF, "true");
 
     DEFAULTS = ImmutableMap.copyOf(defaultSettings);
   }
 
   protected CONFIG config;
 
-  public SchemaGenerator(Map<String, ?> settings) {
+  public AbstractSchemaGenerator(Map<String, ?> settings) {
     Map<String, Object> copySettings = new LinkedHashMap<>(settings);
 
     for (Map.Entry<String, Object> kvp : DEFAULTS.entrySet()) {
@@ -135,7 +135,7 @@ public static void main(String... args) throws Exception {
       }
     }
 
-    final SchemaGenerator generator;
+    final AbstractSchemaGenerator generator;
     final String type = ns.getString("type");
 
     if ("csv".equalsIgnoreCase(type)) {
@@ -152,8 +152,8 @@ public static void main(String... args) throws Exception {
 
     Properties properties = new Properties();
     properties.putAll(settings);
-    properties.setProperty(SpoolDirSourceConnectorConfig.KEY_SCHEMA_CONF, ObjectMapperFactory.INSTANCE.writeValueAsString(kvp.getKey()));
-    properties.setProperty(SpoolDirSourceConnectorConfig.VALUE_SCHEMA_CONF, ObjectMapperFactory.INSTANCE.writeValueAsString(kvp.getValue()));
+    properties.setProperty(AbstractSpoolDirSourceConnectorConfig.KEY_SCHEMA_CONF, ObjectMapperFactory.INSTANCE.writeValueAsString(kvp.getKey()));
+    properties.setProperty(AbstractSpoolDirSourceConnectorConfig.VALUE_SCHEMA_CONF, ObjectMapperFactory.INSTANCE.writeValueAsString(kvp.getValue()));
 
     String output = ns.getString("output");
     final String comment = "Configuration was dynamically generated. Please verify before submitting.";