index @timesstamp field added

Author: vsboldakov

lib/sink/ElasticSinkConnector.js

@@ -23,6 +23,7 @@ class ElasticSinkConnector extends SinkConnector {
             index: this.properties.restSink.index,
             type: this.properties.restSink.type,
             mode: this.properties.restSink.mode,
+            indexTimestampFieldName: this.properties.restSink.indexTimestampFieldName,
         };
 
         callback(null, taskConfig);

lib/sink/ElasticSinkTask.js

@@ -15,13 +15,15 @@ class ElasticSinkTask extends SinkTask {
             index,
             type,
             batchSize,
+            indexTimestampFieldName,
         } = this.properties;
 
         this.esClient = esClient;
         this.idProperty = idProperty;
         this.index = index;
         this.type = type;
         this.batchSize = batchSize;
+        this.indexTimestampFieldName = indexTimestampFieldName;
 
         this.upserts = new Map();
         this.deletes = new Set();
@@ -57,6 +59,9 @@ class ElasticSinkTask extends SinkTask {
                 let id = record[this.idProperty];
                 body.push({ "update": {"_id": id, "_index": this.index}});
                 delete record[this.idProperty];
+                if (this.indexTimestampFieldName && record.doc) {
+                    record.doc[this.indexTimestampFieldName] = (new Date()).getTime();
+                }
                 body.push(record);
             }
         }