You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2021/12/02 11:33:01 UTC
[camel] 02/02: CAMEL-16521: camel-kafka - The OVERRIDE_TIMESTAMP header can now be string or other types that are automatic converted to long type.
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
commit 5a7ddc9563afe86a0e3e1e18e9295f1a632cbcb7
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Dec 2 12:32:19 2021 +0100
CAMEL-16521: camel-kafka - The OVERRIDE_TIMESTAMP header can now be string or other types that are automatic converted to long type.
---
.../java/org/apache/camel/component/kafka/KafkaProducer.java | 4 ++--
.../kafka/producer/support/KeyValueHolderIterator.java | 11 +++++------
.../org/apache/camel/component/kafka/KafkaProducerTest.java | 6 ++++--
3 files changed, 11 insertions(+), 10 deletions(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 97a6e6d..017645e 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -176,9 +176,9 @@ public class KafkaProducer extends DefaultAsyncProducer {
Long timeStamp = null;
Object overrideTimeStamp = message.removeHeader(KafkaConstants.OVERRIDE_TIMESTAMP);
- if (overrideTimeStamp instanceof Long) {
+ if (overrideTimeStamp != null) {
+ timeStamp = exchange.getContext().getTypeConverter().convertTo(Long.class, exchange, overrideTimeStamp);
LOG.debug("Using override TimeStamp: {}", overrideTimeStamp);
- timeStamp = (Long) overrideTimeStamp;
}
// extracting headers which need to be propagated
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
index b46983a..8e37321 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/producer/support/KeyValueHolderIterator.java
@@ -107,13 +107,12 @@ public class KeyValueHolderIterator implements Iterator<KeyValueHolder<Object, P
}
private Long getOverrideTimestamp(Message innerMessage) {
- Object objTimestamp = innerMessage.getHeader(KafkaConstants.OVERRIDE_TIMESTAMP);
-
- if (hasValidTimestampHeader(objTimestamp)) {
- return (Long) innerMessage.removeHeader(KafkaConstants.OVERRIDE_TIMESTAMP);
+ Long timeStamp = null;
+ Object overrideTimeStamp = innerMessage.removeHeader(KafkaConstants.OVERRIDE_TIMESTAMP);
+ if (overrideTimeStamp != null) {
+ timeStamp = exchange.getContext().getTypeConverter().convertTo(Long.class, exchange, overrideTimeStamp);
}
-
- return null;
+ return timeStamp;
}
private String getInnerTopic(Message innerMessage) {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 9495020..ec2750d 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -222,8 +222,10 @@ public class KafkaProducerTest {
in.setHeader(KafkaConstants.PARTITION_KEY, 4);
in.setHeader(KafkaConstants.OVERRIDE_TOPIC, "anotherTopic");
in.setHeader(KafkaConstants.KEY, "someKey");
- in.setHeader(KafkaConstants.OVERRIDE_TIMESTAMP,
- LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
+
+ // test using a string value instead of long
+ String time = "" + LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ in.setHeader(KafkaConstants.OVERRIDE_TIMESTAMP, time);
producer.process(exchange);