review comments

Make TimestampSource a builder
typos and header comments

Author: Mike Skells

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

@@ -202,7 +202,10 @@ public final ZoneId getFilenameTimezone() {
     }
 
     public final TimestampSource getFilenameTimestampSource() {
-        return TimestampSource.forConfiguration(getString(FILE_NAME_TIMESTAMP_SOURCE), getFilenameTimezone());
+        return new TimestampSource.Builder()
+                .configuration(getString(FILE_NAME_TIMESTAMP_SOURCE))
+                .zoneId(getFilenameTimezone())
+                .build();
     }
 
     public final int getMaxRecordsPerFile() {

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2024 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+public final class PathAccess {
+    private final String[] terms;
+
+    private PathAccess(final String[] terms) {
+        this.terms = terms;
+    }
+
+    /**
+     * Parse a path definition string into a Path object. The path definition string is a '.' separated 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 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 For example "a.b.c" would parse into a path with
+     * terms "a", "b", "c" For example ".:a.b:c" would parse into a path with terms "a.b", "c"
+     *
+     * @return a PathAccess that can access a value in a nested structure
+     */
+    public static PathAccess parse(final String pathDefinition) {
+        final String pathDescription;
+        final String pathSeparator;
+        if (pathDefinition.length() > 1 && pathDefinition.charAt(0) == '.' ) {
+            pathDescription = pathDefinition.substring(2);
+            pathSeparator = pathDefinition.substring(1,2);
+        } else {
+            pathDescription = pathDefinition;
+            pathSeparator = ".";
+        }
+        return new PathAccess(Pattern.compile(pathSeparator, Pattern.LITERAL).split(pathDescription));
+    }
+
+    public Object extractDataFrom(final SinkRecord value) {
+        Object current = value;
+
+        for (final String term : terms) {
+            if (current == null) {
+                return null;
+            }
+            if (current instanceof Struct) {
+                final Struct struct = (Struct) current;
+                final Schema schema = struct.schema();
+                final 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;
+    }
+
+    @Override
+    public String toString() {
+        return "PathAccess[terms=" + Arrays.toString( terms) +"]";
+    }
+}

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

@@ -18,7 +18,10 @@
 
 import java.time.Instant;
 import java.time.ZoneId;
+import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
+import java.util.Locale;
+import java.util.Objects;
 
 import org.apache.kafka.connect.header.Header;
 import org.apache.kafka.connect.sink.SinkRecord;
@@ -39,58 +42,67 @@ enum Type {
         CUSTOM
 
     }
+    class Builder {
+        private ZoneId zoneId = ZoneOffset.UTC;
+        private Type type;
+        private String additionalParameters;
 
-    static TimestampSource forConfiguration(final String configuration, final ZoneId zoneId) {
-        final String[] parts = configuration.split(":", 2);
-        final String typeName = parts[0];
-        final String context = parts.length > 1 ? parts[1] : null;
-        for (final Type t : Type.values()) {
-            if (t.name().equalsIgnoreCase(typeName)) {
-                return create(t, context, zoneId);
-            }
+        public Builder zoneId(final ZoneId zoneId) {
+            Objects.requireNonNull(zoneId, "zoneId cannot be null");
+            this.zoneId = zoneId;
+            return this;
         }
-        throw new IllegalArgumentException(String.format("Unknown timestamp source: %s", configuration));
-    }
 
-    private static TimestampSource create(Type t, String context, ZoneId zoneId) {
-        switch (t) {
-            case WALLCLOCK:
-                if (context != null) {
-                    throw new IllegalArgumentException("Wallclock timestamp source does not support context");
-                }
-                return new WallclockTimestampSource(zoneId);
-            case EVENT:
-                if (context != null) {
-                    throw new IllegalArgumentException("Event timestamp source does not support context");
-                }
-                return new EventTimestampSource(zoneId);
-            case DATA:
-                if (context == null) {
-                    throw new IllegalArgumentException("Data timestamp source requires context");
-                }
-                return new DataTimestampSource(context, zoneId);
-            case HEADER:
-                if (context == null) {
-                    throw new IllegalArgumentException("Header timestamp source requires context");
-                }
-                return new HeaderTimestampSource(context, zoneId);
-            case CUSTOM:
-                if (context == null) {
-                    throw new IllegalArgumentException("Header timestamp source requires context");
-                }
-                final String[] parts = context.split(":", 2);
-                final String className = parts[0];
-                final String params = parts.length > 1 ? parts[1] : null;
-                try {
-                    final Class<?> clazz = Class.forName(className);
-                    return (TimestampSource) clazz.getConstructor(String.class, ZoneId.class).newInstance(params, zoneId);
-                } catch (final Exception e) {
-                    throw new IllegalArgumentException("Failed to create custom timestamp source", e);
-                }
-            default:
-                throw new IllegalArgumentException(
-                        String.format("Unsupported timestamp extractor type: %s", t));
+        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));
+
+            this.additionalParameters = parts.length > 1 ? parts[1] : null;
+            return this;
         }
+
+        public TimestampSource build() {
+            switch (type) {
+                case WALLCLOCK:
+                    if (additionalParameters != null) {
+                        throw new IllegalArgumentException("Wallclock timestamp source does not support additionalParameters");
+                    }
+                    return new WallclockTimestampSource(zoneId);
+                case EVENT:
+                    if (additionalParameters != null) {
+                        throw new IllegalArgumentException("Event timestamp source does not support additionalParameters");
+                    }
+                    return new EventTimestampSource(zoneId);
+                case DATA:
+                    if (additionalParameters == null) {
+                        throw new IllegalArgumentException("Data timestamp source requires additionalParameters");
+                    }
+                    return new DataTimestampSource(additionalParameters, zoneId);
+                case HEADER:
+                    if (additionalParameters == null) {
+                        throw new IllegalArgumentException("Header timestamp source requires additionalParameters");
+                    }
+                    return new HeaderTimestampSource(additionalParameters, zoneId);
+                case CUSTOM:
+                    if (additionalParameters == null) {
+                        throw new IllegalArgumentException("Header timestamp source requires additionalParameters");
+                    }
+                    final String[] parts = additionalParameters.split(":", 2);
+                    final String className = parts[0];
+                    final String params = parts.length > 1 ? parts[1] : null;
+                    try {
+                        final Class<?> clazz = Class.forName(className);
+                        return (TimestampSource) clazz.getConstructor(String.class, ZoneId.class).newInstance(params, zoneId);
+                    } catch (final Exception e) {
+                        throw new IllegalArgumentException("Failed to create custom timestamp source", e);
+                    }
+                default:
+                    throw new IllegalArgumentException(
+                            String.format("Unsupported timestamp extractor type: %s", type));
+            }
+        }
+
     }
 
     abstract class AbstractTimestampSource implements TimestampSource {
@@ -148,16 +160,16 @@ public ZonedDateTime time(final SinkRecord record) {
     }
 
     final class DataTimestampSource extends AbstractTimestampSource {
-        private final Path path;
+        private final PathAccess path;
 
         DataTimestampSource(final String pathDefinition, final ZoneId zoneId) {
             super(zoneId, Type.DATA);
-            this.path = Path.parse(pathDefinition);
+            this.path = PathAccess.parse(pathDefinition);
         }
 
         @Override
         public ZonedDateTime time(final SinkRecord record) {
-            Object value = path.extractDataFrom(record);
+            final Object value = path.extractDataFrom(record);
             return rawTime(value);
         }
 
@@ -173,7 +185,7 @@ final class HeaderTimestampSource extends AbstractTimestampSource {
 
         @Override
         public ZonedDateTime time(final SinkRecord record) {
-            Header header = record.headers().lastWithName(headerName);
+            final Header header = record.headers().lastWithName(headerName);
             return rawTime(header == null ? null : header.value());
         }
     }

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

@@ -21,14 +21,14 @@
 
 import io.aiven.kafka.connect.common.config.TimestampSource;
 
-import java.time.ZoneOffset;
-
 public class TimestampSourceValidator implements ConfigDef.Validator {
 
     @Override
     public void ensureValid(final String name, final Object value) {
         try {
-            TimestampSource.forConfiguration(value.toString(), ZoneOffset.UTC);
+            new TimestampSource.Builder()
+                    .configuration(value.toString())
+                    .build();
         } catch (final Exception e) { // NOPMD AvoidCatchingGenericException
             throw new ConfigException(name, value, e.getMessage());
         }