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 4c00fef75223 camel-kafka: Add topicHeaderName option to TimestampRouter
4c00fef75223 is described below
commit 4c00fef752237dfd33ca1578983e3346fbafad04
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Jun 23 09:52:47 2026 +0200
camel-kafka: Add topicHeaderName option to TimestampRouter
Add optional topicHeaderName parameter to TimestampRouter bean, allowing
callers to specify which header contains the topic name instead of always
using KafkaConstants.TOPIC (CamelKafkaTopic). This fixes compatibility
with the timestamp-router-action kamelet when used behind a webhook source,
where Camel's HTTP components filter CamelKafka-prefixed headers.
Closes #24191
Co-Authored-By: Claude <[email protected]>
---
.../component/kafka/transform/TimestampRouter.java | 10 ++-
.../kafka/transform/TimestampRouterTest.java | 95 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 2 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/TimestampRouter.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/TimestampRouter.java
index 5c8e4fe42ea8..d7f0e9c0cc57 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/TimestampRouter.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/transform/TimestampRouter.java
@@ -32,7 +32,8 @@ public class TimestampRouter {
public void process(
@ExchangeProperty("topicFormat") String topicFormat,
@ExchangeProperty("timestampFormat") String timestampFormat,
- @ExchangeProperty("timestampHeaderName") String
timestampHeaderName, Exchange ex) {
+ @ExchangeProperty("timestampHeaderName") String
timestampHeaderName,
+ @ExchangeProperty("topicHeaderName") String topicHeaderName,
Exchange ex) {
final Pattern TOPIC = Pattern.compile("$[topic]", Pattern.LITERAL);
final Pattern TIMESTAMP = Pattern.compile("$[timestamp]",
Pattern.LITERAL);
@@ -41,7 +42,12 @@ public class TimestampRouter {
fmt.setTimeZone(TimeZone.getTimeZone("UTC"));
Long timestamp = null;
- String topicName = ex.getMessage().getHeader(KafkaConstants.TOPIC,
String.class);
+ String topicName;
+ if (ObjectHelper.isNotEmpty(topicHeaderName)) {
+ topicName = ex.getMessage().getHeader(topicHeaderName,
String.class);
+ } else {
+ topicName = ex.getMessage().getHeader(KafkaConstants.TOPIC,
String.class);
+ }
Object rawTimestamp = ex.getMessage().getHeader(timestampHeaderName);
if (rawTimestamp instanceof Long longValue) {
timestamp = longValue;
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/TimestampRouterTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/TimestampRouterTest.java
new file mode 100644
index 000000000000..6bcfa4d391e0
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/transform/TimestampRouterTest.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.transform;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class TimestampRouterTest {
+
+ private final TimestampRouter router = new TimestampRouter();
+
+ @Test
+ public void testDefaultTopicHeader() {
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC, "my-topic");
+ exchange.getMessage().setHeader("kafka.TIMESTAMP", "1719100800000");
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd",
"kafka.TIMESTAMP", null, exchange);
+
+ assertEquals("my-topic_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ public void testCustomTopicHeaderName() {
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+ exchange.getMessage().setHeader("kafka.TOPIC", "my-topic-3");
+ exchange.getMessage().setHeader("kafka.TIMESTAMP", "1719100800000");
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd",
"kafka.TIMESTAMP", "kafka.TOPIC", exchange);
+
+ assertEquals("my-topic-3_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ public void testEmptyTopicHeaderNameFallsBackToDefault() {
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC,
"fallback-topic");
+ exchange.getMessage().setHeader("kafka.TIMESTAMP", "1719100800000");
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd",
"kafka.TIMESTAMP", "", exchange);
+
+ assertEquals("fallback-topic_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ public void testNullTopicHeaderNameFallsBackToDefault() {
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC,
"fallback-topic");
+ exchange.getMessage().setHeader("kafka.TIMESTAMP", "1719100800000");
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd",
"kafka.TIMESTAMP", null, exchange);
+
+ assertEquals("fallback-topic_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ public void testNoTopicHeaderSet() {
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+ exchange.getMessage().setHeader("kafka.TIMESTAMP", "1719100800000");
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd",
"kafka.TIMESTAMP", null, exchange);
+
+ assertEquals("_2024-06-23",
exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+
+ @Test
+ public void testNoTimestamp() {
+ Exchange exchange = new DefaultExchange(new DefaultCamelContext());
+ exchange.getMessage().setHeader(KafkaConstants.TOPIC, "my-topic");
+
+ router.process("$[topic]_$[timestamp]", "yyyy-MM-dd",
"kafka.TIMESTAMP", null, exchange);
+
+
assertNull(exchange.getMessage().getHeader(KafkaConstants.OVERRIDE_TOPIC));
+ }
+}
