This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new 7cd2ceafe2 Fix #1292: set the topic, partition, offset and timestamp
headers on sink exchanges (#1862)
7cd2ceafe2 is described below
commit 7cd2ceafe2cdf459088446be0888c6fce57eb181
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 11:37:30 2026 +0100
Fix #1292: set the topic, partition, offset and timestamp headers on sink
exchanges (#1862)
The sink exchange only carried the record key, as the
camel.kafka.connector.record.key header, so a sink route could not use
the record's position in Kafka, as a route consuming from Kafka with
Camel can.
- Set camel.kafka.connector.record.topic, .partition, .offset and
.timestamp (when the record has one) on each sink exchange. They are
set after the CamelHeader.* record headers are mapped, so a record
header cannot replace them.
- Document the headers set by a sink connector, with an example that
uses them in a file name.
- Test the headers, the precedence over record headers, and the Simple
expression used in the docs.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/kafkaconnector/CamelSinkTask.java | 16 +++++++++
.../camel/kafkaconnector/CamelSinkTaskTest.java | 40 ++++++++++++++++++++++
.../ROOT/pages/user-guide/basic-configuration.adoc | 19 ++++++++++
3 files changed, 75 insertions(+)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
index 5ade2ccf72..cacbc3969c 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -48,6 +48,10 @@ public class CamelSinkTask extends SinkTask {
public static final String KAMELET_SINK_TEMPLATE_PARAMETERS_PREFIX =
"camel.kamelet.ckcSink.";
public static final String KAFKA_RECORD_KEY_HEADER =
"camel.kafka.connector.record.key";
+ public static final String KAFKA_RECORD_TOPIC_HEADER =
"camel.kafka.connector.record.topic";
+ public static final String KAFKA_RECORD_PARTITION_HEADER =
"camel.kafka.connector.record.partition";
+ public static final String KAFKA_RECORD_OFFSET_HEADER =
"camel.kafka.connector.record.offset";
+ public static final String KAFKA_RECORD_TIMESTAMP_HEADER =
"camel.kafka.connector.record.timestamp";
public static final String HEADER_CAMEL_PREFIX = "CamelHeader.";
public static final String PROPERTY_CAMEL_PREFIX = "CamelProperty.";
@@ -226,6 +230,9 @@ public class CamelSinkTask extends SinkTask {
}
}
}
+ // set after the record headers are mapped, so that a
CamelHeader.* record header cannot replace the
+ // coordinates of the record in Kafka
+ setRecordMetadataHeaders(exchange, record);
LOG.debug("Sending exchange {} to {}", exchange.getExchangeId(),
LOCAL_URL);
producer.send(localEndpoint, exchange);
@@ -313,6 +320,15 @@ public class CamelSinkTask extends SinkTask {
}
}
+ private static void setRecordMetadataHeaders(Exchange exchange, SinkRecord
record) {
+ exchange.getMessage().setHeader(KAFKA_RECORD_TOPIC_HEADER,
record.topic());
+ exchange.getMessage().setHeader(KAFKA_RECORD_PARTITION_HEADER,
record.kafkaPartition());
+ exchange.getMessage().setHeader(KAFKA_RECORD_OFFSET_HEADER,
record.kafkaOffset());
+ if (record.timestamp() != null) {
+ exchange.getMessage().setHeader(KAFKA_RECORD_TIMESTAMP_HEADER,
record.timestamp());
+ }
+ }
+
private static void mapHeader(Header header, String prefix, Map<String,
Object> destination) {
final String key = StringHelper.after(header.key(), prefix,
header.key());
final Schema schema = header.schema();
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
index 26c76171ea..be77a123bd 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
@@ -30,6 +30,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
@@ -90,6 +91,45 @@ class CamelSinkTaskTest {
.getString(CamelSinkConnectorConfig.CAMEL_SINK_CONTENT_LOG_LEVEL_CONF));
}
+ @Test
+ void testRecordMetadataHeaders() {
+ Map<String, String> props = new HashMap<>();
+ props.put(TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
+
+ sinkTask.start(props);
+
+ List<SinkRecord> records = new ArrayList<SinkRecord>();
+ SinkRecord record = new SinkRecord(TOPIC_NAME, 3, null, "test", null,
"camel", 42L, 1234567890L, TimestampType.CREATE_TIME);
+ // record headers must not be able to replace the coordinates of the
record
+ record.headers().addString(CamelSinkTask.HEADER_CAMEL_PREFIX +
CamelSinkTask.KAFKA_RECORD_TOPIC_HEADER, "spoofed");
+ record.headers().addLong(CamelSinkTask.HEADER_CAMEL_PREFIX +
CamelSinkTask.KAFKA_RECORD_OFFSET_HEADER, 7L);
+ records.add(record);
+ // a record without a timestamp
+ records.add(new SinkRecord(TOPIC_NAME, 4, null, "test", null, "camel",
43));
+ sinkTask.put(records);
+
+ ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
+ Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
+ Map<String, Object> headers = exchange.getMessage().getHeaders();
+ assertEquals("test",
headers.get(CamelSinkTask.KAFKA_RECORD_KEY_HEADER));
+ assertEquals(TOPIC_NAME,
headers.get(CamelSinkTask.KAFKA_RECORD_TOPIC_HEADER));
+ assertEquals(3,
headers.get(CamelSinkTask.KAFKA_RECORD_PARTITION_HEADER));
+ assertEquals(42L,
headers.get(CamelSinkTask.KAFKA_RECORD_OFFSET_HEADER));
+ assertEquals(1234567890L,
headers.get(CamelSinkTask.KAFKA_RECORD_TIMESTAMP_HEADER));
+ // the expression shown in the docs (basic-configuration.adoc)
+ assertEquals(TOPIC_NAME + "-3-42.txt",
exchange.getContext().resolveLanguage("simple")
+
.createExpression("${header[camel.kafka.connector.record.topic]}-${header[camel.kafka.connector.record.partition]}"
+ +
"-${header[camel.kafka.connector.record.offset]}.txt")
+ .evaluate(exchange, String.class));
+
+ Exchange exchange1 = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
+ Map<String, Object> headers1 = exchange1.getMessage().getHeaders();
+ assertEquals(4,
headers1.get(CamelSinkTask.KAFKA_RECORD_PARTITION_HEADER));
+ assertEquals(43L,
headers1.get(CamelSinkTask.KAFKA_RECORD_OFFSET_HEADER));
+
assertFalse(headers1.containsKey(CamelSinkTask.KAFKA_RECORD_TIMESTAMP_HEADER));
+ }
+
@Test
void testTopicsRegex() {
Map<String, String> props = new HashMap<>();
diff --git a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
index c257210167..f9927db862 100644
--- a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
+++ b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
@@ -87,6 +87,25 @@ transforms.toString.target.type=java.lang.String
[[BasicConfiguration-MessageLevelHeaders]]
== Message Level Headers
+A sink connector sets these headers on each Camel message it sends:
+
+[width="100%",cols="2,5",options="header"]
+|===
+| Header | Value
+| camel.kafka.connector.record.key | The key of the record
+| camel.kafka.connector.record.topic | The topic of the record
+| camel.kafka.connector.record.partition | The partition of the record, as an
`Integer`
+| camel.kafka.connector.record.offset | The offset of the record, as a `Long`
+| camel.kafka.connector.record.timestamp | The timestamp of the record in
milliseconds, as a `Long`, when the record has one
+|===
+
+The topic, partition, offset and timestamp are set after the record headers
described below are mapped, so a `CamelHeader.` record header cannot replace
them. The topic is the one the record has when it reaches the connector, after
any SMT that changes it. Header names that contain dots are written in brackets
in the Camel Simple language. For example, `camel-file-kafka-connector` can
write each record to its own file, named after the record's position in Kafka:
+
+[source,properties]
+----
+camel.sink.endpoint.fileName=${header[camel.kafka.connector.record.topic]}-${header[camel.kafka.connector.record.partition]}-${header[camel.kafka.connector.record.offset]}.txt
+----
+
Certain sink components offer additional configuration and adjustment of
behavior by setting specific headers on the message sent via Kafka. For example
the file sink connector allows the user to set the file name either via the
configuration property or via the header `CamelHeader.CamelFileName`. This is
usually noted on the connector documentation, but as a rule of thumb, users are
advised to also check the Camel components documentation to verify what message
headers are available.
If they are available, to configure them directly from your Kafka producer,
you can append them with the string `CamelHeader.` (i.e.:
`CamelHeader.CamelFileName`, `CamelHeader.CamelFilePath`, etc) and include them
on the message headers from Kafka.
