Add ExtractTimestamp transformation and tests

Author: ivanyu

README.md

@@ -6,9 +6,34 @@ This is a set of [Kafka Connect transformations](https://kafka.apache.org/docume
 
 ## Transformations
 
+See [the Kafka documentation](https://kafka.apache.org/documentation/#connect_transforms) for more details about configuring transformations. 
+
+### `ExtractTimestamp`
+
+This transformation replaces the original record's timestamp with a value taken from the the record.
+
+The transformation:
+- expects the record value to be either a `STRUCT` or a `MAP`;
+- expects it to have a specified field;
+- expects the value of the field to be either `INT64` or `org.apache.kafka.connect.data.Timestamp` and not be `null`.
+
+Exists in one variant:
+- `io.aiven.kafka.connect.transforms.ExtractTimestamp$Value` - works on values.
+
+The transformation defines the following configurations:
+
+- `field.name` - The name of the field which should be used as the new timestamp. Cannot be `null` or empty.
+
+Here's an example of this transformation configuration:
+
+```properties
+transforms=ExtractTimestampFromValueField
+transforms.ExtractTimestampFromValueField.field.name=inner_field_name
+```
+
 ### `ExtractTopic`
 
-This transformation extracts a string value from the message and use it as the topic name.
+This transformation extracts a string value from the record and use it as the topic name.
 
 The transformation can use either the whole key or value (in this case, it must have `STRING` type) or a field in them (in this case, it must have `STRUCT` type and the field's value must be `STRING`).
 
@@ -28,8 +53,6 @@ transforms=ExtractTopicFromValueField
 transforms.ExtractTopicFromValueField.field.name=inner_field_name
 ```
 
-See [the Kafka documentation](https://kafka.apache.org/documentation/#connect_transforms) for more details about configuring transformations. 
-
 ## License
 
 This project is licensed under the [Apache License, Version 2.0](LICENSE).

src/main/java/io/aiven/kafka/connect/transforms/ExtractTimestamp.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2019 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.transforms;
+
+import java.util.Date;
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.transforms.Transformation;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ExtractTimestamp<R extends ConnectRecord<R>> implements Transformation<R> {
+    private static final Logger log = LoggerFactory.getLogger(ExtractTimestamp.class);
+
+    private ExtractTimestampConfig config;
+
+    @Override
+    public ConfigDef config() {
+        return ExtractTimestampConfig.config();
+    }
+
+    @Override
+    public void configure(final Map<String, ?> configs) {
+        this.config = new ExtractTimestampConfig(configs);
+    }
+
+    @Override
+    public R apply(final R record) {
+        if (record.value() == null) {
+            throw new DataException("Value can't be null: " + record);
+        }
+
+        final Object fieldValue;
+        if (record.value() instanceof Struct) {
+            final Struct struct = (Struct) record.value();
+            if (struct.schema().field(config.fieldName()) == null) {
+                throw new DataException(config.fieldName() + " field must be present and its value can't be null: "
+                    + record);
+            }
+            fieldValue = struct.get(config.fieldName());
+        } else if (record.value() instanceof Map) {
+            final Map map = (Map) record.value();
+            fieldValue = map.get(config.fieldName());
+        } else {
+            throw new DataException("Value type must be STRUCT or MAP: " + record);
+        }
+
+        if (fieldValue == null) {
+            throw new DataException(config.fieldName() + " field must be present and its value can't be null: "
+                + record);
+        }
+
+        final long newTimestamp;
+        if (fieldValue instanceof Long) {
+            newTimestamp = (long) fieldValue;
+        } else if (fieldValue instanceof Date) {
+            newTimestamp = ((Date) fieldValue).getTime();
+        } else {
+            throw new DataException(config.fieldName()
+                + " field must be INT64 or org.apache.kafka.connect.data.Timestamp: "
+                + record);
+        }
+
+        return record.newRecord(
+            record.topic(),
+            record.kafkaPartition(),
+            record.keySchema(),
+            record.key(),
+            record.valueSchema(),
+            record.value(),
+            newTimestamp
+        );
+    }
+
+    public static final class Value<R extends ConnectRecord<R>> extends ExtractTimestamp<R> {
+        // There's an implementation only for value, not for key.
+        // We provide $Value class anyway for the consistency sake
+        // and in case we need a $Key version in the future as well.
+    }
+
+    @Override
+    public void close() {
+    }
+}

src/main/java/io/aiven/kafka/connect/transforms/ExtractTimestampConfig.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2019 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.transforms;
+
+import java.util.Map;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+final class ExtractTimestampConfig extends AbstractConfig {
+    public static final String FIELD_NAME_CONFIG = "field.name";
+    private static final String FIELD_NAME_DOC =
+        "The name of the field is to be used as the source of timestamp. "
+            + "The field must have INT64 or org.apache.kafka.connect.data.Timestamp type "
+            + "and must mot be null.";
+
+    ExtractTimestampConfig(final Map<?, ?> originals) {
+        super(config(), originals);
+    }
+
+    static ConfigDef config() {
+        return new ConfigDef()
+            .define(
+                FIELD_NAME_CONFIG,
+                ConfigDef.Type.STRING,
+                ConfigDef.NO_DEFAULT_VALUE,
+                new ConfigDef.NonEmptyString(),
+                ConfigDef.Importance.HIGH,
+                FIELD_NAME_DOC);
+    }
+
+    final String fieldName() {
+        return getString(FIELD_NAME_CONFIG);
+    }
+}

src/test/java/io/aiven/kafka/connect/transforms/ExtractTimestampConfigTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 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.transforms;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigException;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ExtractTimestampConfigTest {
+    @Test
+    void emptyConfig() {
+        final Map<String, String> props = new HashMap<>();
+        final Throwable e = assertThrows(ConfigException.class, () -> new ExtractTimestampConfig(props));
+        assertEquals("Missing required configuration \"field.name\" which has no default value.",
+            e.getMessage());
+    }
+
+    @Test
+    void emptyFieldName() {
+        final Map<String, String> props = new HashMap<>();
+        props.put("field.name", "");
+        final Throwable e = assertThrows(ConfigException.class, () -> new ExtractTimestampConfig(props));
+        assertEquals("Invalid value  for configuration field.name: String must be non-empty",
+            e.getMessage());
+    }
+
+    @Test
+    void definedFieldName() {
+        final Map<String, String> props = new HashMap<>();
+        props.put("field.name", "test");
+        final ExtractTimestampConfig config = new ExtractTimestampConfig(props);
+        assertEquals("test", config.fieldName());
+    }
+}