Add optional timestamp argument to AddMetadataConverter (#143)

awslabs · Oct 8, 2018 · a52ac1a · a52ac1a
1 parent 6bde3f7
commit a52ac1a
Showing 1 changed file with 15 additions and 1 deletion.
diff --git a/src/com/amazon/kinesis/streaming/agent/processing/processors/AddMetadataConverter.java b/src/com/amazon/kinesis/streaming/agent/processing/processors/AddMetadataConverter.java
@@ -15,8 +15,12 @@
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
 import java.util.Map;
 import java.util.LinkedHashMap;
+import java.util.TimeZone;
 
 import com.amazon.kinesis.streaming.agent.ByteBuffers;
 import com.amazon.kinesis.streaming.agent.config.Configuration;
@@ -33,6 +37,7 @@
  *
  * { 
  *   "optionName": "ADDMETADATA",
+ *   "timestamp": "true/false",
  *   "metadata": {
  *     "key": "value",
  *     "foo": {
@@ -47,10 +52,12 @@
 public class AddMetadataConverter implements IDataConverter {
 
     private Object metadata;
+    private Boolean timestamp;
     private final IJSONPrinter jsonProducer;
 
     public AddMetadataConverter(Configuration config) {
       metadata = config.getConfigMap().get("metadata");
+      timestamp = new Boolean((String) config.getConfigMap().get("timestamp"));
       jsonProducer = ProcessingUtilsFactory.getPrinter(config);
     }
 
@@ -64,6 +71,13 @@ public ByteBuffer convert(ByteBuffer data) throws DataConversionException {
             dataStr = dataStr.substring(0, (dataStr.length() - NEW_LINE.length()));
         }
 
+        if (timestamp.booleanValue()) {
+            TimeZone tz = TimeZone.getTimeZone("UTC");
+            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+            dateFormat.setTimeZone(tz);
+            recordMap.put("ts", dateFormat.format(new Date()));
+        }
+
         recordMap.put("metadata", metadata);
         recordMap.put("data", dataStr);
 
@@ -75,4 +89,4 @@ public ByteBuffer convert(ByteBuffer data) throws DataConversionException {
     public String toString() {
         return getClass().getSimpleName();
     }
-}
+}