feat(plugin) add new offset-strategy to use unix-inode

streamthoughts · Aug 3, 2020 · 7345103 · 7345103
1 parent 1b5869d
commit 7345103
Showing 1 changed file with 15 additions and 0 deletions.
diff --git a/...plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/offset/OffsetStrategy.java b/...plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/offset/OffsetStrategy.java
@@ -18,6 +18,7 @@
  */
 package io.streamthoughts.kafka.connect.filepulse.offset;
 
+import io.streamthoughts.kafka.connect.filepulse.errors.ConnectFilePulseException;
 import io.streamthoughts.kafka.connect.filepulse.source.SourceMetadata;
 
 import java.util.Collections;
@@ -50,10 +51,24 @@ Map<String, Object> toPartitionMap(final SourceMetadata metadata) {
         Map<String, Object> toPartitionMap(final SourceMetadata metadata) {
             return Collections.singletonMap(FILEPATH_FIELD, metadata.path());
         }
+    },
+
+    INODE("inode") {
+        @Override
+        Map<String, Object> toPartitionMap(final SourceMetadata metadata) {
+            if (metadata.inode() == null) {
+                throw new ConnectFilePulseException(
+                    "Unix-inode is not supported. " +
+                    "Consider configuring a different 'offset-strategy' (i.e: path, name, hash, name+hash)");
+            }
+            return Collections.singletonMap(INODE_FIELD, metadata.inode());
+        }
     };
 
+
     private static final String FILEPATH_FIELD = "path";
     private static final String FILENAME_FIELD = "name";
+    private static final String INODE_FIELD    = "inode";
     private static final String CRC32_FIELD    = "hash";
 
     private final String label;