allow a more complex path

Author: Mike Skells

commons/src/main/java/io/aiven/kafka/connect/common/config/Path.java

@@ -0,0 +1,72 @@
+package io.aiven.kafka.connect.common.config;
+
+
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+import java.util.List;
+import java.util.Map;
+
+public class Path {
+    private final String[] terms;
+
+    public Path(String[] terms) {
+        this.terms = terms;
+    }
+
+    /**
+     * Parse a path definition string into a Path object.
+     * The path definition string is a '.' seperated series of strings, which are the terms in the path
+     * If the '.' is something that should be included in the terms, and you want to use a different seperator, then
+     * you can specify the seperator as the first character in the descriptions, and have the second character as ':'
+     * @param pathDefinition
+     * @return
+     */
+    public static Path parse(String pathDefinition) {
+        final String pathDescription;
+                final String pathSeperator;
+        if (pathDefinition.length()> 1 && pathDefinition.charAt(1) == ':' ) {
+            pathDescription = pathDefinition.substring(2);
+            pathSeperator = pathDefinition.substring(0,1);
+        } else {
+            pathDescription = pathDefinition;
+            pathSeperator = ":";
+        }
+        return new Path(pathDescription.split(pathSeperator));
+        
+    }
+
+    public Object extractDataFrom(SinkRecord value) {
+        Object current = value;
+        //Schema scurrentSchema = value.valueSchema();
+
+        for (String term : terms) {
+            if (current == null) {
+                return null;
+            }
+            if (current instanceof Struct) {
+                Struct struct = (Struct) current;
+                Schema schema = struct.schema();
+                Field field = schema.field(term);
+                if (field == null) {
+                    return null;
+                }
+                current = struct.get(field);
+            }
+            if (current instanceof Map) {
+                current = ((Map<?, ?>) current).get(term);
+            } else if (current instanceof List) {
+                try {
+                    current = ((List<?>) current).get(Integer.parseInt(term));
+                } catch (NumberFormatException|IndexOutOfBoundsException e) {
+                    return null;
+                }
+            } else {
+                return null;
+            }
+        }
+        return current;
+    }
+}

commons/src/main/java/io/aiven/kafka/connect/common/config/TimestampSource.java

@@ -20,9 +20,6 @@
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 
-import org.apache.kafka.connect.data.Field;
-import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.header.Header;
 import org.apache.kafka.connect.sink.SinkRecord;
 
@@ -151,28 +148,19 @@ public ZonedDateTime time(final SinkRecord record) {
     }
 
     final class DataTimestampSource extends AbstractTimestampSource {
-        private final String fieldName;
+        private final Path path;
 
-        DataTimestampSource(final String field, final ZoneId zoneId) {
+        DataTimestampSource(final String pathDefinition, final ZoneId zoneId) {
             super(zoneId, Type.DATA);
-            this.fieldName = field;
+            this.path = Path.parse(pathDefinition);
         }
 
         @Override
         public ZonedDateTime time(final SinkRecord record) {
-            Object value = extract(record.value());
+            Object value = path.extractDataFrom(record);
             return rawTime(value);
         }
 
-        private Object extract(Object value) {
-            if (value instanceof Struct) {
-                Struct data = (Struct) value;
-                Schema schema = data.schema();
-                Field field = schema.field(fieldName);
-                return field == null ? null : data.get(field);
-            }
-            return null;
-        }
     }
 
     final class HeaderTimestampSource extends AbstractTimestampSource {