Allow TimestampNowField to output TTL for DynamoDB

.editorconfig

@@ -0,0 +1,4 @@
+root = true
+[*.java]
+indent_style = space
+indent_size = 2

src/main/java/com/github/jcustenborder/kafka/connect/transform/common/TimestampNowField.java

@@ -27,15 +27,14 @@
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.data.Struct;
-import org.apache.kafka.connect.data.Timestamp;
 
 import java.util.Collection;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+
 @Title("TimestampNowField")
 @Description("This transformation is used to set a field with the current timestamp of the system running the " +
     "transformation.")
@@ -66,25 +65,20 @@ public void close() {
 
   Map<Schema, Schema> schemaCache = new HashMap<>();
 
-  static boolean isTimestampSchema(Schema schema) {
-    return (Timestamp.SCHEMA.type() == schema.type() && Timestamp.SCHEMA.name().equals(schema.name()));
-  }
-
   @Override
   protected SchemaAndValue processStruct(R record, Schema inputSchema, Struct input) {
-    Date timestamp = new Date(this.time.milliseconds());
-
+    Object timestamp = getFormattedTimestamp();
     Schema outputSchema = schemaCache.computeIfAbsent(inputSchema, schema -> {
       Collection<String> replaceFields = schema.fields().stream()
           .filter(f -> this.config.fields.contains(f.name()))
-          .filter(f -> !isTimestampSchema(f.schema()))
+          .filter(f -> !this.config.targetType.isMatchingSchema(f.schema()))
           .map(Field::name)
           .collect(Collectors.toList());
       SchemaBuilder builder = SchemaBuilders.of(schema, replaceFields);
       this.config.fields.forEach(timestampField -> {
         Field existingField = builder.field(timestampField);
         if (null == existingField) {
-          builder.field(timestampField, Timestamp.SCHEMA);
+          builder.field(timestampField, this.config.targetType.getSchema());
         }
       });
       return builder.build();
@@ -98,10 +92,18 @@ protected SchemaAndValue processStruct(R record, Schema inputSchema, Struct inpu
     return new SchemaAndValue(outputSchema, output);
   }
 
+  private Object getFormattedTimestamp() {
+    long desiredTimeInMillis = this.time.milliseconds();
+    if (config.addAmount > 0) {
+      desiredTimeInMillis += config.addAmount * config.addChronoUnit.getDuration().toMillis();
+    }
+    return this.config.targetType.getFormattedTimestamp(desiredTimeInMillis);
+  }
+
   @Override
   protected SchemaAndValue processMap(R record, Map<String, Object> input) {
     Map<String, Object> result = new LinkedHashMap<>(input);
-    Date timestamp = new Date(this.time.milliseconds());
+    Object timestamp = getFormattedTimestamp();
     this.config.fields.forEach(field -> result.put(field, timestamp));
     return new SchemaAndValue(null, result);
   }

src/main/java/com/github/jcustenborder/kafka/connect/transform/common/TimestampNowFieldConfig.java

@@ -17,21 +17,42 @@
 
 import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder;
 import com.github.jcustenborder.kafka.connect.utils.config.ConfigUtils;
+import com.github.jcustenborder.kafka.connect.utils.config.validators.ValidChronoUnit;
+import com.github.jcustenborder.kafka.connect.utils.config.validators.Validators;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 
+import java.time.temporal.ChronoUnit;
 import java.util.Map;
 import java.util.Set;
 
 class TimestampNowFieldConfig extends AbstractConfig {
   public static final String FIELDS_CONF = "fields";
+  public static final String ADD_AMOUNT_CONF = "add.amount";
+  public static final String ADD_CHRONO_UNIT_CONF = "add.chronounit";
+
+  public static final String TARGET_TYPE_CONF = "target.type";
+
   public static final String FIELDS_DOC = "The field(s) that will be inserted with the timestamp of the system.";
 
+  public static final String ADD_AMOUNT_DOC = "how many of the chosen ChronoUnits to add on top of the timestamp of the system.";
+
+  public static final String ADD_CHRONO_UNIT_DOC = "String representation of the ChronoUnit to add, eg: 'DAYS'";
+
+  public static final String TARGET_TYPE_DOC = "The desired timestamp representation: Unix, Date";
+
   public final Set<String> fields;
+  public final Long addAmount;
+  public final ChronoUnit addChronoUnit;
+
+  public final TimestampNowFieldTargetType targetType;
 
   public TimestampNowFieldConfig(Map<?, ?> originals) {
     super(config(), originals);
     this.fields = ConfigUtils.getSet(this, FIELDS_CONF);
+    this.addAmount = getLong(ADD_AMOUNT_CONF);
+    this.addChronoUnit = ChronoUnit.valueOf(getString(ADD_CHRONO_UNIT_CONF));
+    this.targetType = ConfigUtils.getEnum(TimestampNowFieldTargetType.class, this, TARGET_TYPE_CONF);
   }
 
   public static ConfigDef config() {
@@ -41,6 +62,26 @@ public static ConfigDef config() {
                 .documentation(FIELDS_DOC)
                 .importance(ConfigDef.Importance.HIGH)
                 .build()
+        ).define(
+            ConfigKeyBuilder.of(ADD_AMOUNT_CONF, ConfigDef.Type.LONG)
+                .documentation(ADD_AMOUNT_DOC)
+                .importance(ConfigDef.Importance.LOW)
+                .defaultValue("0")
+                .build()
+        ).define(
+            ConfigKeyBuilder.of(ADD_CHRONO_UNIT_CONF, ConfigDef.Type.STRING)
+                .documentation(ADD_CHRONO_UNIT_DOC)
+                .importance(ConfigDef.Importance.LOW)
+                .defaultValue("DAYS")
+                .validator(new ValidChronoUnit())
+                .build()
+        ).define(
+            ConfigKeyBuilder.of(TARGET_TYPE_CONF, ConfigDef.Type.STRING)
+                .documentation(TARGET_TYPE_DOC)
+                .importance(ConfigDef.Importance.LOW)
+                .defaultValue("DATE")
+                .validator(Validators.validEnum(TimestampNowFieldTargetType.class))
+                .build()
         );
   }
 }

src/main/java/com/github/jcustenborder/kafka/connect/transform/common/TimestampNowFieldTargetType.java

@@ -0,0 +1,63 @@
+/**
+ * Copyright © 2017 Jeremy Custenborder ([email protected])
+ * <p>
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.github.jcustenborder.kafka.connect.transform.common;
+
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Timestamp;
+
+import java.time.Instant;
+import java.util.Date;
+
+public enum TimestampNowFieldTargetType {
+  DATE {
+    @Override
+    boolean isMatchingSchema(Schema schema) {
+      return Timestamp.SCHEMA.type() == getSchema().type() && Timestamp.SCHEMA.name().equals(schema.name());
+    }
+
+    @Override
+    Schema getSchema() {
+      return Timestamp.SCHEMA;
+    }
+
+    @Override
+    Object getFormattedTimestamp(long timeInMillis) {
+      return new Date(timeInMillis);
+    }
+  },
+  UNIX {
+    @Override
+    boolean isMatchingSchema(Schema schema) {
+      return Schema.Type.INT64 == schema.type() && null == schema.name();
+    }
+
+    @Override
+    Schema getSchema() {
+      return Schema.INT64_SCHEMA;
+    }
+
+    @Override
+    Object getFormattedTimestamp(long timeInMillis) {
+      return Instant.ofEpochMilli(timeInMillis).getEpochSecond();
+    }
+  };
+
+  abstract boolean isMatchingSchema(Schema schema);
+
+  abstract Schema getSchema();
+
+  abstract Object getFormattedTimestamp(long timeInMillis);
+}

src/main/java/com/github/jcustenborder/kafka/connect/utils/config/validators/ValidChronoUnit.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright © 2017 Jeremy Custenborder ([email protected])
+ * <p>
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.github.jcustenborder.kafka.connect.utils.config.validators;
+
+import com.google.common.base.Joiner;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class ValidChronoUnit implements ConfigDef.Validator {
+  List<String> validOptions = Arrays.stream(ChronoUnit.values()).map(ChronoUnit::name).collect(Collectors.toList());
+
+  @Override
+  public void ensureValid(String s, Object o) {
+    if (o instanceof String) {
+      if (!validOptions.contains(o)) {
+        throw new ConfigException(
+          s,
+          String.format(
+            "'%s' is not a valid value for %s. Valid values are %s.",
+            o,
+            ChronoUnit.class.getSimpleName(),
+            Joiner.on(", ").join(validOptions)
+          )
+        );
+      }
+    } else if (o instanceof List) {
+      List list = (List) o;
+      for (Object i : list) {
+        ensureValid(s, i);
+      }
+    } else {
+      throw new ConfigException(
+        s,
+        o,
+        "Must be a String or List"
+      );
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "Matches: ``" + Joiner.on("``, ``").join(this.validOptions) + "``";
+  }
+}