fix #738 : added timestamp information to source records.

connectors/camel-timer-kafka-connector/src/main/java/org/apache/camel/kafkaconnector/timer/CamelTimerSourceTask.java

@@ -16,13 +16,14 @@
  */
 package org.apache.camel.kafkaconnector.timer;
 
-import java.util.HashMap;
-import java.util.Map;
-import javax.annotation.Generated;
+import org.apache.camel.Exchange;
 import org.apache.camel.kafkaconnector.CamelSourceConnectorConfig;
 import org.apache.camel.kafkaconnector.CamelSourceTask;
 
-@Generated("This class has been generated by camel-kafka-connector-generator-maven-plugin, remove this annotation to prevent it from being generated.")
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
 public class CamelTimerSourceTask extends CamelSourceTask {
 
     @Override
@@ -36,4 +37,17 @@ protected Map<String, String> getDefaultConfig() {
             put(CamelSourceConnectorConfig.CAMEL_SOURCE_COMPONENT_CONF, "timer");
         }};
     }
+
+    //XXX: this method override is the only difference from how the class was initially generated by camel-kafka-connector-generator-maven-plugin
+    @Override
+    protected long calculateTimestamp(Exchange exchange) {
+        if (exchange != null) {
+            Date fireDate = exchange.getProperty(Exchange.TIMER_FIRED_TIME, Date.class);
+            if (fireDate != null) {
+                return fireDate.getTime();
+            }
+        }
+
+        return super.calculateTimestamp(exchange);
+    }
 }

core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java

@@ -173,9 +173,11 @@ public synchronized List<SourceRecord> poll() {
             final Schema messageKeySchema = messageHeaderKey != null ? SchemaHelper.buildSchemaBuilderForType(messageHeaderKey) : null;
             final Schema messageBodySchema = messageBodyValue != null ? SchemaHelper.buildSchemaBuilderForType(messageBodyValue) : null;
 
+            final long timestamp = calculateTimestamp(exchange);
+
             for (String singleTopic : topics) {
-                SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, singleTopic, messageKeySchema,
-                        messageHeaderKey, messageBodySchema, messageBodyValue);
+                SourceRecord record = new SourceRecord(sourcePartition, sourceOffset, singleTopic, null, messageKeySchema,
+                        messageHeaderKey, messageBodySchema, messageBodyValue, timestamp);
 
                 if (exchange.getMessage().hasHeaders()) {
                     setAdditionalHeaders(record, exchange.getMessage().getHeaders(), HEADER_CAMEL_PREFIX);
@@ -240,6 +242,10 @@ protected static String getCamelSourcePathConfigPrefix() {
         return CAMEL_SOURCE_PATH_PROPERTIES_PREFIX;
     }
 
+    protected long calculateTimestamp(Exchange exchange) {
+        return System.currentTimeMillis();
+    }
+
     private void setAdditionalHeaders(SourceRecord record, Map<String, Object> map, String prefix) {
 
         for (Map.Entry<String, Object> entry : map.entrySet()) {

core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java

@@ -37,6 +37,7 @@
 import static org.apache.camel.util.CollectionHelper.mapOf;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -67,6 +68,7 @@ public void testSourcePolling() {
 
         assertEquals(size, poll.size());
         assertEquals(TOPIC_NAME, poll.get(0).topic());
+        assertNotNull(poll.get(0).timestamp());
         assertEquals(LoggingLevel.OFF.toString(), sourceTask.getCamelSourceConnectorConfig(props)
             .getString(CamelSourceConnectorConfig.CAMEL_SOURCE_CONTENT_LOG_LEVEL_CONF));