Added check for cleanup policy. Added log message when removing file.

jcustenborder · May 2, 2018 · e6c6ec4 · e6c6ec4
1 parent 3af4d58
commit e6c6ec4
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/src/main/java/com/github/jcustenborder/kafka/connect/spooldir/SpoolDirSourceTask.java b/src/main/java/com/github/jcustenborder/kafka/connect/spooldir/SpoolDirSourceTask.java
@@ -132,9 +132,12 @@ public void start(Map<String, String> settings) {
     this.config = config(settings);
 
     checkDirectory(SpoolDirSourceConnectorConfig.INPUT_PATH_CONFIG, this.config.inputPath);
-    checkDirectory(SpoolDirSourceConnectorConfig.FINISHED_PATH_CONFIG, this.config.finishedPath);
     checkDirectory(SpoolDirSourceConnectorConfig.ERROR_PATH_CONFIG, this.config.errorPath);
 
+    if (SpoolDirSourceConnectorConfig.CleanupPolicy.MOVE == this.config.cleanupPolicy) {
+      checkDirectory(SpoolDirSourceConnectorConfig.FINISHED_PATH_CONFIG, this.config.finishedPath);
+    }
+
     this.parser = new Parser();
     Map<Schema, TypeParser> dateTypeParsers = ImmutableMap.of(
         Timestamp.SCHEMA, new TimestampTypeParser(this.config.parserTimestampTimezone, this.config.parserTimestampDateFormats),
@@ -288,6 +291,7 @@ private void closeAndDelete() throws IOException {
       log.info("Closing {}", this.inputFile);
       this.inputStream.close();
       this.inputStream = null;
+      log.info("Removing file {}", this.inputFile);
       this.inputFile.delete();
       File processingFile = InputFileDequeue.processingFile(this.config.processingFileExtension, this.inputFile);
       if (processingFile.exists()) {