review comments

Remove a few simple classes and make a DataExtractor to read things from the `sinkRecord`
and few tidyups

Author: Mike Skells

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

@@ -23,7 +23,9 @@
 import java.util.Locale;
 import java.util.Objects;
 
-import org.apache.kafka.connect.header.Header;
+import io.aiven.kafka.connect.common.config.extractors.DataExtractor;
+import io.aiven.kafka.connect.common.config.extractors.HeaderValueExtractor;
+import io.aiven.kafka.connect.common.config.extractors.SimpleValuePath;
 import org.apache.kafka.connect.sink.SinkRecord;
 
 public interface TimestampSource {
@@ -38,7 +40,7 @@ enum Type {
         WALLCLOCK,
         EVENT,
         HEADER,
-        DATA,
+        SIMPLE_DATA,
         CUSTOM
 
     }
@@ -47,16 +49,56 @@ class Builder {
         private Type type;
         private String additionalParameters;
 
+        /**
+         * set the zoneId to be used. If this method isnt called, the default is UTC
+         * @return this
+         * @throws NullPointerException if zoneId is null
+         */
         public Builder zoneId(final ZoneId zoneId) {
             Objects.requireNonNull(zoneId, "zoneId cannot be null");
             this.zoneId = zoneId;
             return this;
         }
 
+        /**
+         * sets the type of the timestamp source and associated parameters (if needed)
+         * The format of the configuration is <type>[:<data>]
+         * i.e. the type name, optionally followed by data.
+         * <br>
+         * The data is type specific
+         * <p>
+         * For type WALLCLOCK or EVENT, no data is allowed
+         * </p>
+         * <p>
+         * For type SIMPLE_DATA, data is required, and is a '.' separated series of
+         * terms in the path
+         * <br>If the '.' is something that should be included in the terms, and you
+         * want to use a different separator, then you can specify a '.' as the first character, and the separator as the
+         * second character, and then the path is the rest of the string
+         * <br>For example "SIMPLE_DATA:a.b.c" would use into a path with
+         * terms "a", "b", "c"
+         * <br>For example "SIMPLE_DATA:.:a.b:c" would use a path with terms "a.b", "c"
+         * </p>
+         * For type HEADER, data is required, and is the name of the header to extract
+         * <br>For example "HEADER:foo" would use to "foo" header (or null if its not available in the SinkRecord
+         * </p>
+         * </p>
+         * For type CUSTOM, data is required, and is the name of the class to use, and any additional parameters for that custom time source.
+         * The specified class must implement the TimestampSource interface and have a public constructor that takes a String and a ZoneId. Fort the meaning of the data, see the documentation of the custom class.
+         * <br>For example "CUSTOM:my.custom.timesource:some more data" would be similar to calling new my.custom.timesource("some more data", zoneId)
+         * </p>
+         *
+
+         * @return this
+         */
         public Builder configuration(final String configuration) {
             final String[] parts = configuration.split(":", 2);
             final String typeName = parts[0];
-            this.type = Type.valueOf(typeName.toUpperCase(Locale.ENGLISH));
+            try {
+                this.type = Type.valueOf(typeName.toUpperCase(Locale.ENGLISH));
+            } catch (final IllegalArgumentException e) {
+                throw new IllegalArgumentException("Unknown timestamp source: "+typeName);
+            }
 
             this.additionalParameters = parts.length > 1 ? parts[1] : null;
             return this;
@@ -74,16 +116,16 @@ public TimestampSource build() {
                         throw new IllegalArgumentException("Event timestamp source does not support additionalParameters");
                     }
                     return new EventTimestampSource(zoneId);
-                case DATA:
+                case SIMPLE_DATA:
                     if (additionalParameters == null) {
                         throw new IllegalArgumentException("Data timestamp source requires additionalParameters");
                     }
-                    return new DataTimestampSource(additionalParameters, zoneId);
+                    return new SimpleTimestampSource(zoneId, Type.SIMPLE_DATA, SimpleValuePath.parse(additionalParameters));
                 case HEADER:
                     if (additionalParameters == null) {
                         throw new IllegalArgumentException("Header timestamp source requires additionalParameters");
                     }
-                    return new HeaderTimestampSource(additionalParameters, zoneId);
+                    return new SimpleTimestampSource(zoneId, Type.HEADER, new HeaderValueExtractor(additionalParameters));
                 case CUSTOM:
                     if (additionalParameters == null) {
                         throw new IllegalArgumentException("Header timestamp source requires additionalParameters");
@@ -105,41 +147,48 @@ public TimestampSource build() {
 
     }
 
-    abstract class AbstractTimestampSource implements TimestampSource {
+    class SimpleTimestampSource implements TimestampSource {
         protected final ZoneId zoneId;
         private final Type type;
+        private final DataExtractor dataExtractor;
 
-        protected AbstractTimestampSource(final ZoneId zoneId, final Type type) {
+        protected SimpleTimestampSource(final ZoneId zoneId, final Type type, DataExtractor dataExtractor) {
             this.zoneId = zoneId;
             this.type = type;
+            this.dataExtractor = dataExtractor;
         }
 
         @Override
         public Type type() {
             return type;
         }
 
-        protected ZonedDateTime rawTime(final Object rawValue) {
+        @Override
+        public ZonedDateTime time(SinkRecord record) {
+            return fromRawTime(dataExtractor.extractDataFrom(record));
+        }
+
+        protected ZonedDateTime fromRawTime(final Object rawValue) {
             if (rawValue == null) {
                 return null;
             } else if (rawValue instanceof Long) {
-                return time((Long) rawValue);
+                return withZone((Long) rawValue);
             } else if (rawValue instanceof ZonedDateTime) {
                 return (ZonedDateTime) rawValue;
             } else if (rawValue instanceof Instant) {
-                return time(((Instant) rawValue).toEpochMilli());
+                return withZone(((Instant) rawValue).toEpochMilli());
             }
             return null;
         }
 
-        protected ZonedDateTime time(final long timestamp) {
+        protected ZonedDateTime withZone(final long timestamp) {
             return ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestamp), zoneId);
         }
     }
 
-    final class WallclockTimestampSource extends AbstractTimestampSource {
+    final class WallclockTimestampSource extends SimpleTimestampSource {
         WallclockTimestampSource(final ZoneId zoneId) {
-            super(zoneId, Type.WALLCLOCK);
+            super(zoneId, Type.WALLCLOCK, null);
         }
 
         @Override
@@ -148,45 +197,14 @@ public ZonedDateTime time(final SinkRecord record) {
         }
     }
 
-    final class EventTimestampSource extends AbstractTimestampSource {
+    final class EventTimestampSource extends SimpleTimestampSource {
         EventTimestampSource(final ZoneId zoneId) {
-            super(zoneId, Type.EVENT);
-        }
-
-        @Override
-        public ZonedDateTime time(final SinkRecord record) {
-            return time(record.timestamp());
-        }
-    }
-
-    final class DataTimestampSource extends AbstractTimestampSource {
-        private final PathAccess path;
-
-        DataTimestampSource(final String pathDefinition, final ZoneId zoneId) {
-            super(zoneId, Type.DATA);
-            this.path = PathAccess.parse(pathDefinition);
-        }
-
-        @Override
-        public ZonedDateTime time(final SinkRecord record) {
-            final Object value = path.extractDataFrom(record);
-            return rawTime(value);
-        }
-
-    }
-
-    final class HeaderTimestampSource extends AbstractTimestampSource {
-        private final String headerName;
-
-        HeaderTimestampSource(final String headerName, final ZoneId zoneId) {
-            super(zoneId, Type.HEADER);
-            this.headerName = headerName;
+            super(zoneId, Type.EVENT, null);
         }
 
         @Override
         public ZonedDateTime time(final SinkRecord record) {
-            final Header header = record.headers().lastWithName(headerName);
-            return rawTime(header == null ? null : header.value());
+            return withZone(record.timestamp());
         }
     }
 }

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

@@ -0,0 +1,8 @@
+package io.aiven.kafka.connect.common.config.extractors;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+
+public interface DataExtractor {
+
+    Object extractDataFrom(final SinkRecord record);
+}

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

@@ -0,0 +1,17 @@
+package io.aiven.kafka.connect.common.config.extractors;
+
+import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+public class HeaderValueExtractor implements DataExtractor {
+    private final String headerKey;
+
+    public HeaderValueExtractor(final String headerKey) {
+        this.headerKey = headerKey;
+    }
+
+    public Object extractDataFrom(final SinkRecord record) {
+        final Header header = record.headers().lastWithName(headerKey);
+        return header == null ? null : header.value();
+    }
+}

commons/src/main/java/io/aiven/kafka/connect/common/config/PathAccess.java renamed to commons/src/main/java/io/aiven/kafka/connect/common/config/extractors/SimpleValuePath.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.aiven.kafka.connect.common.config;
+package io.aiven.kafka.connect.common.config.extractors;
 
 
 import org.apache.kafka.connect.data.Field;
@@ -26,10 +26,10 @@
 import java.util.Map;
 import java.util.regex.Pattern;
 
-public final class PathAccess {
+public final class SimpleValuePath implements DataExtractor {
     private final String[] terms;
 
-    private PathAccess(final String[] terms) {
+    private SimpleValuePath(final String[] terms) {
         this.terms = terms;
     }
 
@@ -42,7 +42,7 @@ private PathAccess(final String[] terms) {
      *
      * @return a PathAccess that can access a value in a nested structure
      */
-    public static PathAccess parse(final String pathDefinition) {
+    public static SimpleValuePath parse(final String pathDefinition) {
         final String pathDescription;
         final String pathSeparator;
         if (pathDefinition.length() > 1 && pathDefinition.charAt(0) == '.' ) {
@@ -52,11 +52,11 @@ public static PathAccess parse(final String pathDefinition) {
             pathDescription = pathDefinition;
             pathSeparator = ".";
         }
-        return new PathAccess(Pattern.compile(pathSeparator, Pattern.LITERAL).split(pathDescription));
+        return new SimpleValuePath(Pattern.compile(pathSeparator, Pattern.LITERAL).split(pathDescription));
     }
 
-    public Object extractDataFrom(final SinkRecord value) {
-        Object current = value;
+    public Object extractDataFrom(final SinkRecord record) {
+        Object current = record.value();
 
         for (final String term : terms) {
             if (current == null) {
@@ -70,8 +70,7 @@ public Object extractDataFrom(final SinkRecord value) {
                     return null;
                 }
                 current = struct.get(field);
-            }
-            if (current instanceof Map) {
+            } else if (current instanceof Map) {
                 current = ((Map<?, ?>) current).get(term);
             } else if (current instanceof List) {
                 try {
@@ -88,6 +87,6 @@ public Object extractDataFrom(final SinkRecord value) {
 
     @Override
     public String toString() {
-        return "PathAccess[terms=" + Arrays.toString( terms) +"]";
+        return "Path[terms=" + Arrays.toString( terms) +"]";
     }
 }