[camel] branch main updated: CAMEL-16627: camel-core - Add common header for source timestamp

[davsclaus]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new bd1808f  CAMEL-16627: camel-core - Add common header for source 
timestamp
bd1808f is described below

commit bd1808f4e1418e60fb13e1bb4d9e1e913c45612b
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Wed Jun 9 13:37:30 2021 +0200

    CAMEL-16627: camel-core - Add common header for source timestamp
---
 .../org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java   | 4 ++++
 .../java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java     | 6 +++++-
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git 
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
 
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index 58d083f..3446612 100644
--- 
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++ 
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -185,6 +185,10 @@ public class Kinesis2Consumer extends 
ScheduledBatchPollingConsumer {
         exchange.getIn().setHeader(Kinesis2Constants.APPROX_ARRIVAL_TIME, 
record.approximateArrivalTimestamp());
         exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, 
record.partitionKey());
         exchange.getIn().setHeader(Kinesis2Constants.SEQUENCE_NUMBER, 
record.sequenceNumber());
+        if (record.approximateArrivalTimestamp() != null) {
+            long ts = record.approximateArrivalTimestamp().getEpochSecond() * 
1000;
+            exchange.getIn().setHeader(Exchange.MESSAGE_TIMESTAMP, ts);
+        }
         return exchange;
     }
 
diff --git 
a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
 
b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
index 2c5192e..60e116f 100644
--- 
a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
+++ 
b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
@@ -392,7 +392,6 @@ public class AWS2S3Consumer extends 
ScheduledBatchPollingConsumer {
         message.setHeader(AWS2S3Constants.KEY, key);
         message.setHeader(AWS2S3Constants.BUCKET_NAME, 
getConfiguration().getBucketName());
         message.setHeader(AWS2S3Constants.E_TAG, s3Object.response().eTag());
-        message.setHeader(AWS2S3Constants.LAST_MODIFIED, 
s3Object.response().lastModified());
         message.setHeader(AWS2S3Constants.VERSION_ID, 
s3Object.response().versionId());
         message.setHeader(AWS2S3Constants.CONTENT_TYPE, 
s3Object.response().contentType());
         message.setHeader(AWS2S3Constants.CONTENT_LENGTH, 
s3Object.response().contentLength());
@@ -404,6 +403,11 @@ public class AWS2S3Consumer extends 
ScheduledBatchPollingConsumer {
         message.setHeader(AWS2S3Constants.REPLICATION_STATUS, 
s3Object.response().replicationStatus());
         message.setHeader(AWS2S3Constants.STORAGE_CLASS, 
s3Object.response().storageClass());
         message.setHeader(AWS2S3Constants.METADATA, 
s3Object.response().metadata());
+        if (s3Object.response().lastModified() != null) {
+            message.setHeader(AWS2S3Constants.LAST_MODIFIED, 
s3Object.response().lastModified());
+            long ts = s3Object.response().lastModified().getEpochSecond() * 
1000;
+            message.setHeader(Exchange.MESSAGE_TIMESTAMP, ts);
+        }
 
         /*
          * If includeBody == true, it is safe to close the object here because 
the S3Object