Add additional timestamp sources

.github/workflows/codeql-analysis.yml

@@ -47,6 +47,11 @@ jobs:
         uses: github/codeql-action/init@v3
         with:
           languages: ${{ matrix.language }}
+    # Initializes the CodeQL tools for scanning.
+      - name: Initialize CodeQL
+        uses: github/codeql-action/init@v2
+        with:
+          languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
         # By default, queries listed here will override any specified in a config file.
         # Prefix the list here with "+" to use these queries and those in the config file.

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

@@ -202,8 +202,10 @@ public final ZoneId getFilenameTimezone() {
     }
 
     public final TimestampSource getFilenameTimestampSource() {
-        return TimestampSource.of(getFilenameTimezone(),
-                TimestampSource.Type.of(getString(FILE_NAME_TIMESTAMP_SOURCE)));
+        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/TimestampSource.java

@@ -20,82 +20,199 @@
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
+import java.util.Date;
+import java.util.Locale;
+import java.util.Objects;
 
+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 {
 
     ZonedDateTime time(SinkRecord record);
 
-    @SuppressWarnings("PMD.ShortMethodName")
-    static TimestampSource of(final Type extractorType) {
-        return of(ZoneOffset.UTC, extractorType);
-    }
-
-    @SuppressWarnings("PMD.ShortMethodName")
-    static TimestampSource of(final ZoneId zoneId, final Type extractorType) {
-        switch (extractorType) {
-            case WALLCLOCK :
-                return new WallclockTimestampSource(zoneId);
-            case EVENT :
-                return new EventTimestampSource(zoneId);
-            default :
-                throw new IllegalArgumentException(
-                        String.format("Unsupported timestamp extractor type: %s", extractorType));
-        }
-    }
 
     Type type();
 
     enum Type {
 
-        WALLCLOCK, EVENT;
+        /** Use the current wallclock time */
+        WALLCLOCK,
+        /** Use the event timestamp */
+        EVENT,
+        /** Use a header value to specify the date */
+        HEADER,
+        /** Extract the timestamp from the payload */
+        SIMPLE_DATA,
+        /** Use a custom timestamp source */
+        CUSTOM
+
+    }
+    class Builder {
+        private ZoneId zoneId = ZoneOffset.UTC;
+        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;
+        }
 
-        @SuppressWarnings("PMD.ShortMethodName")
-        public static Type of(final String name) {
-            for (final Type t : Type.values()) {
-                if (t.name().equalsIgnoreCase(name)) {
-                    return t;
-                }
+        /**
+         * 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];
+            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;
+        }
+
+        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 SIMPLE_DATA:
+                    if (additionalParameters == null) {
+                        throw new IllegalArgumentException("Data timestamp source requires additionalParameters");
+                    }
+                    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 SimpleTimestampSource(zoneId, Type.HEADER, new HeaderValueExtractor(additionalParameters));
+                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));
             }
-            throw new IllegalArgumentException(String.format("Unknown timestamp source: %s", name));
         }
 
     }
 
-    final class WallclockTimestampSource implements TimestampSource {
-        private final ZoneId zoneId;
+    class SimpleTimestampSource implements TimestampSource {
+        protected final ZoneId zoneId;
+        private final Type type;
+        private final DataExtractor dataExtractor;
 
-        protected WallclockTimestampSource(final ZoneId zoneId) {
+        protected SimpleTimestampSource(final ZoneId zoneId, final Type type, DataExtractor dataExtractor) {
             this.zoneId = zoneId;
+            this.type = type;
+            this.dataExtractor = dataExtractor;
         }
 
         @Override
-        public ZonedDateTime time(final SinkRecord record) {
-            return ZonedDateTime.now(zoneId);
+        public Type type() {
+            return type;
         }
 
         @Override
-        public Type type() {
-            return Type.WALLCLOCK;
+        public ZonedDateTime time(SinkRecord record) {
+            return fromRawTime(dataExtractor.extractDataFrom(record));
         }
-    }
 
-    final class EventTimestampSource implements TimestampSource {
-        private final ZoneId zoneId;
+        protected ZonedDateTime fromRawTime(final Object rawValue) {
+            if (rawValue == null) {
+                return null;
+            } else if (rawValue instanceof Long) {
+                return withZone((Long) rawValue);
+            } else if (rawValue instanceof Date) {
+                return withZone(((Date) rawValue).getTime());
+            } else if (rawValue instanceof ZonedDateTime) {
+                return (ZonedDateTime) rawValue;
+            } else if (rawValue instanceof Instant) {
+                return withZone(((Instant) rawValue).toEpochMilli());
+            }
+            return null;
+        }
 
-        protected EventTimestampSource(final ZoneId zoneId) {
-            this.zoneId = zoneId;
+        protected ZonedDateTime withZone(final long timestamp) {
+            return ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestamp), zoneId);
+        }
+    }
+
+    final class WallclockTimestampSource extends SimpleTimestampSource {
+        WallclockTimestampSource(final ZoneId zoneId) {
+            super(zoneId, Type.WALLCLOCK, null);
         }
 
         @Override
         public ZonedDateTime time(final SinkRecord record) {
-            return ZonedDateTime.ofInstant(Instant.ofEpochMilli(record.timestamp()), zoneId);
+            return ZonedDateTime.now(zoneId);
+        }
+    }
+
+    final class EventTimestampSource extends SimpleTimestampSource {
+        EventTimestampSource(final ZoneId zoneId) {
+            super(zoneId, Type.EVENT, null);
         }
 
         @Override
-        public Type type() {
-            return Type.EVENT;
+        public ZonedDateTime time(final SinkRecord record) {
+            return withZone(record.timestamp());
         }
     }
 }

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2020 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.extractors;
+
+import org.apache.kafka.connect.sink.SinkRecord;
+
+/**
+ * Extracts data from a {@link SinkRecord}. The actual data extracted is implementation specific
+ */
+public interface DataExtractor {
+
+    /**
+     * Extracts data from a {@link SinkRecord}.
+     *
+     * @param record the record to extract data from
+     * @return the extracted data
+     */
+    Object extractDataFrom(final SinkRecord record);
+}

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 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.extractors;
+
+import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+/**
+ * A DataExtractor that extracts the value of a header from a SinkRecord
+ * If there are multiple headers with the same ane, then the last occurrence will be used
+ * If the header does not exist, then the extraction will yield null
+ */
+public class HeaderValueExtractor implements DataExtractor {
+    private final String headerKey;
+
+    /**
+     * Create a new HeaderValueExtractor
+     *
+     * @param headerKey the key of the header to extract.
+     */
+    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();
+    }
+}