This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 8c6c672b3c [Feature][Connector-V2][Kafka] Support reading Kafka
message headers in source connector (#11521)
8c6c672b3c is described below
commit 8c6c672b3cae310c352c3a14ba729856f797b189
Author: Hyunkyung, Kim <[email protected]>
AuthorDate: Sat Jul 25 16:57:02 2026 +0900
[Feature][Connector-V2][Kafka] Support reading Kafka message headers in
source connector (#11521)
Co-authored-by: Claude Sonnet 4.6 (1M context) <[email protected]>
---
docs/en/connectors/source/Kafka.md | 29 +++
docs/zh/connectors/source/Kafka.md | 30 +++
.../seatunnel/kafka/config/KafkaSourceOptions.java | 10 +
.../seatunnel/kafka/source/ConsumerMetadata.java | 3 +
.../source/KafkaHeadersDeserializationSchema.java | 131 +++++++++++
.../seatunnel/kafka/source/KafkaRecordEmitter.java | 74 +++++-
.../seatunnel/kafka/source/KafkaSourceConfig.java | 80 ++++++-
.../seatunnel/kafka/source/KafkaSourceFactory.java | 3 +-
.../KafkaHeadersDeserializationSchemaTest.java | 254 +++++++++++++++++++++
.../kafka/source/KafkaRecordEmitterTest.java | 105 +++++++++
.../kafka/source/KafkaSourceConfigTest.java | 116 ++++++++++
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 58 +++++
.../resources/kafka/kafka_source_with_headers.conf | 94 ++++++++
13 files changed, 976 insertions(+), 11 deletions(-)
diff --git a/docs/en/connectors/source/Kafka.md
b/docs/en/connectors/source/Kafka.md
index 7402b4e6c2..ac74411b56 100644
--- a/docs/en/connectors/source/Kafka.md
+++ b/docs/en/connectors/source/Kafka.md
@@ -63,6 +63,8 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| protobuf_schema | String
| No | - |
Effective when the format is set to protobuf, specifies the Schema definition
[...]
| strip_schema_registry_header | Boolean
| No | false |
Effective when the format is set to protobuf. Whether to strip the Confluent
Schema Registry wire format header (magic byte, schema id and message indexes)
before protobuf deserialization. This option is useful when consuming Protobuf
messages that were encoded using Confluent Schema Registry. When enabled, the
connector will try to de [...]
| reader_cache_queue_size | Integer
| No | 2 |
The capacity of the fetcher-to-reader element queue. Each element is one
`consumer.poll()` batch, not a single message. See
[reader_cache_queue_size](#reader_cache_queue_size) for details. |
+| is_native | Boolean
| No | false |
Supports retaining the source information of the record.
[...]
+| kafka_headers_fields | Array
| No | - |
Specify which Kafka message header keys to extract as row fields. Each header
value is read as a STRING type and appended to the output row after the regular
schema fields. Cannot be used with NATIVE format.
[...]
> On restore from checkpoint or savepoint, Kafka Source resumes from the
> checkpointed split offsets.
> `start_mode` and consumer-group offsets are only used for the first startup
> or for newly
@@ -509,8 +511,35 @@ source {
```
**Note**: When `strip_schema_registry_header` is enabled, the connector can
safely handle both Schema Registry encoded messages and plain Protobuf
messages. If the Schema Registry header is not detected, it will automatically
fall back to standard Protobuf deserialization.
+
+### Reading Kafka Headers
+
+Use `kafka_headers_fields` to extract specific Kafka message headers as row
fields. The header values are appended as STRING type fields after all regular
schema fields.
+
+> Note: Cannot be used with `NATIVE` format, which already exposes headers as
a `Map<String, String>` field.
+
+```hocon
+source {
+ Kafka {
+ topic = "my-topic"
+ bootstrap.servers = "localhost:9092"
+ kafka_headers_fields = ["correlation-id", "x-trace-id"]
+ schema = {
+ fields {
+ user_id = "int"
+ name = "string"
+ }
+ }
+ format = json
+ }
+}
```
+The output row will contain: `user_id` (int), `name` (string),
`correlation-id` (string), `x-trace-id` (string).
+If a header key is absent in a record, the corresponding field value will be
`null`.
+
+This is the counterpart of `kafka_headers_fields` in the Kafka sink connector,
allowing round-trip header propagation between topics.
+
### Ignore No Leader Partition
When dealing with Kafka clusters that may have temporary leadership issues,
you can configure the connector to ignore partitions without a leader:
diff --git a/docs/zh/connectors/source/Kafka.md
b/docs/zh/connectors/source/Kafka.md
index 89dfd2da13..36d423b24c 100644
--- a/docs/zh/connectors/source/Kafka.md
+++ b/docs/zh/connectors/source/Kafka.md
@@ -63,6 +63,8 @@ import ChangeLog from '../changelog/connector-kafka.md';
| protobuf_schema | String |
否 | - | 当格式设置为 protobuf 时有效，指定 Schema 定义。
|
| strip_schema_registry_header | Boolean |
否 | false | 当格式设置为 protobuf 时有效。是否在 Protobuf 反序列化之前去除
Confluent Schema Registry 线格式头部（magic byte、schema id 和 message indexes）。当消费使用
Confluent Schema Registry 编码的 Protobuf 消息时，此选项非常有用。启用后，连接器将尝试在解析 Protobuf
消息之前检测并删除 Schema Registry 头部。如果未检测到头部，它将回退到标准的 Protobuf 反序列化。
[...]
| reader_cache_queue_size | Integer |
否 | 2 | Fetcher 与 Reader 线程之间缓冲队列的容量。每个元素是一次
`consumer.poll()` 的整批结果，而非单条消息。详见
[reader_cache_queue_size](#reader_cache_queue_size)。 |
+| is_native | Boolean |
否 | false | 支持保留record的源信息。
|
+| kafka_headers_fields | Array |
否 | - | 指定要从 Kafka 消息 header 中提取并映射为行字段的 header
key 列表。每个 header 值以 STRING 类型追加到输出行的末尾（位于正常 schema 字段之后）。不支持 NATIVE 格式。
|
> 从 checkpoint 或 savepoint 恢复时，Kafka Source 会优先使用 checkpoint 中保存的 split offset。
> `start_mode` 和 consumer group offset 只在首次启动，或为尚未存在 checkpoint
> 状态的新发现分区初始化位点时生效。
@@ -136,6 +138,34 @@ transform {
## 任务示例
+### 读取 Kafka 消息 Header
+
+使用 `kafka_headers_fields` 将指定的 Kafka 消息 header 提取为行字段。header 值以 STRING 类型追加在正常
schema 字段之后。
+
+> 注意：不支持 `NATIVE` 格式，该格式已通过 `Map<String, String>` 字段暴露了所有 header。
+
+```hocon
+source {
+ Kafka {
+ topic = "my-topic"
+ bootstrap.servers = "localhost:9092"
+ kafka_headers_fields = ["correlation-id", "x-trace-id"]
+ schema = {
+ fields {
+ user_id = "int"
+ name = "string"
+ }
+ }
+ format = json
+ }
+}
+```
+
+输出行将包含：`user_id`（int）、`name`（string）、`correlation-id`（string）、`x-trace-id`（string）。
+如果某条消息中不存在对应的 header key，则该字段值为 `null`。
+
+此功能与 Kafka sink 连接器的 `kafka_headers_fields` 对应，支持 header 在 topic 间的全链路传递。
+
### 简单示例
> 此示例读取 Kafka 的 topic_1、topic_2 和 topic_3 的数据并将其打印到客户端。如果尚未安装和部署 SeaTunnel，请按照
> [安装指南](../../getting-started/locally/deployment.md) 进行安装和部署。然后，按照
> [快速开始](../../getting-started/locally/quick-start-seatunnel-engine.md) 运行此任务。
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
index f89ee28cc6..0766e71b42 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.shade.com.fasterxml.jackson.core.type.TypeReference;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;
+import java.util.List;
import java.util.Map;
public class KafkaSourceOptions extends KafkaBaseOptions {
@@ -146,4 +147,13 @@ public class KafkaSourceOptions extends KafkaBaseOptions {
"Effective when the format is avro. The writer
Avro schema used to "
+ "deserialize binary Avro messages whose
record name, namespace, "
+ "or union layout differs from the
SeaTunnel schema.");
+
+ public static final Option<List<String>> KAFKA_HEADERS_FIELDS =
+ Options.key("kafka_headers_fields")
+ .listType()
+ .noDefaultValue()
+ .withDescription(
+ "Specify which Kafka message header keys to
extract as row fields. "
+ + "Each header value is read as a STRING
and appended to the output row. "
+ + "Cannot be used with NATIVE format.");
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/ConsumerMetadata.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/ConsumerMetadata.java
index 91591253ce..5df76f039e 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/ConsumerMetadata.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/ConsumerMetadata.java
@@ -27,6 +27,8 @@ import org.apache.kafka.common.TopicPartition;
import lombok.Data;
import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -43,4 +45,5 @@ public class ConsumerMetadata implements Serializable {
private DeserializationSchema<SeaTunnelRow> deserializationSchema;
private CatalogTable catalogTable;
private Long endOffsetsTimestamp;
+ private List<String> kafkaHeaderFields = Collections.emptyList();
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaHeadersDeserializationSchema.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaHeadersDeserializationSchema.java
new file mode 100644
index 0000000000..0d7eab1344
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaHeadersDeserializationSchema.java
@@ -0,0 +1,131 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.kafka.source;
+
+import org.apache.seatunnel.api.serialization.DeserializationSchema;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/**
+ * A {@link DeserializationSchema} wrapper that appends Kafka message header
values as additional
+ * fields to the deserialized {@link SeaTunnelRow}.
+ *
+ * <p>The headers for the current record are provided via {@link
#setCurrentRecordHeaders(Headers)}
+ * before deserialization is invoked.
+ */
+public class KafkaHeadersDeserializationSchema implements
DeserializationSchema<SeaTunnelRow> {
+
+ private final DeserializationSchema<SeaTunnelRow> delegate;
+ private final List<String> headerFieldNames;
+ private final SeaTunnelRowType extendedRowType;
+ private final int baseFieldCount;
+
+ private Headers currentRecordHeaders;
+
+ public KafkaHeadersDeserializationSchema(
+ DeserializationSchema<SeaTunnelRow> delegate,
+ List<String> headerFieldNames,
+ SeaTunnelRowType extendedRowType) {
+ this.delegate = delegate;
+ this.headerFieldNames = headerFieldNames;
+ this.extendedRowType = extendedRowType;
+ this.baseFieldCount = ((SeaTunnelRowType)
delegate.getProducedType()).getTotalFields();
+ }
+
+ public void setCurrentRecordHeaders(Headers headers) {
+ this.currentRecordHeaders = headers;
+ }
+
+ @Override
+ public SeaTunnelRow deserialize(byte[] message) throws IOException {
+ SeaTunnelRow baseRow = delegate.deserialize(message);
+ return appendHeaderFields(baseRow);
+ }
+
+ @Override
+ public void deserialize(byte[] message, Collector<SeaTunnelRow> out)
throws IOException {
+ delegate.deserialize(
+ message,
+ new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow record) {
+ out.collect(appendHeaderFields(record));
+ }
+
+ @Override
+ public void collect(SchemaChangeEvent event) {
+ out.collect(event);
+ }
+
+ @Override
+ public void markSchemaChangeBeforeCheckpoint() {
+ out.markSchemaChangeBeforeCheckpoint();
+ }
+
+ @Override
+ public void markSchemaChangeAfterCheckpoint() {
+ out.markSchemaChangeAfterCheckpoint();
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return out.getCheckpointLock();
+ }
+ });
+ }
+
+ @Override
+ public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
+ return extendedRowType;
+ }
+
+ private SeaTunnelRow appendHeaderFields(SeaTunnelRow baseRow) {
+ if (baseRow == null) {
+ return null;
+ }
+ Object[] fields = new Object[baseFieldCount + headerFieldNames.size()];
+ System.arraycopy(baseRow.getFields(), 0, fields, 0, baseFieldCount);
+
+ for (int i = 0; i < headerFieldNames.size(); i++) {
+ String headerKey = headerFieldNames.get(i);
+ if (currentRecordHeaders != null) {
+ Header header = currentRecordHeaders.lastHeader(headerKey);
+ fields[baseFieldCount + i] =
+ header != null && header.value() != null
+ ? new String(header.value(),
StandardCharsets.UTF_8)
+ : null;
+ }
+ }
+
+ SeaTunnelRow newRow = new SeaTunnelRow(fields);
+ newRow.setRowKind(baseRow.getRowKind());
+ newRow.setTableId(baseRow.getTableId());
+ newRow.setOptions(baseRow.getOptions());
+ return newRow;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java
index 66810cf33c..576bab45b7 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java
@@ -28,10 +28,14 @@ import
org.apache.seatunnel.format.compatible.kafka.connect.json.CompatibleKafka
import
org.apache.seatunnel.format.compatible.kafka.connect.json.NativeKafkaConnectDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
import java.util.Map;
public class KafkaRecordEmitter
@@ -59,16 +63,30 @@ public class KafkaRecordEmitter
throws Exception {
outputCollector.output = collector;
// todo there is an additional loss in this place for non-multi-table
scenarios
+ ConsumerMetadata consumerMetadata =
mapMetadata.get(splitState.getTablePath());
DeserializationSchema<SeaTunnelRow> deserializationSchema =
-
mapMetadata.get(splitState.getTablePath()).getDeserializationSchema();
+ consumerMetadata.getDeserializationSchema();
if (deserializationSchema instanceof
KafkaEventTimeDeserializationSchema) {
- ((KafkaEventTimeDeserializationSchema) deserializationSchema)
- .setCurrentRecordTimestamp(consumerRecord.timestamp());
+ KafkaEventTimeDeserializationSchema eventTimeSchema =
+ (KafkaEventTimeDeserializationSchema)
deserializationSchema;
+
eventTimeSchema.setCurrentRecordTimestamp(consumerRecord.timestamp());
+ if (eventTimeSchema.getDelegate() instanceof
KafkaHeadersDeserializationSchema) {
+ ((KafkaHeadersDeserializationSchema)
eventTimeSchema.getDelegate())
+ .setCurrentRecordHeaders(consumerRecord.headers());
+ }
}
try {
if (deserializationSchema instanceof
CompatibleKafkaConnectDeserializationSchema) {
+ List<String> kafkaHeaderFields =
consumerMetadata.getKafkaHeaderFields();
+ Collector<SeaTunnelRow> targetCollector =
+ kafkaHeaderFields.isEmpty()
+ ? outputCollector
+ : headerInjectingCollector(
+ outputCollector,
+ consumerRecord.headers(),
+ kafkaHeaderFields);
((CompatibleKafkaConnectDeserializationSchema)
deserializationSchema)
- .deserialize(consumerRecord, outputCollector);
+ .deserialize(consumerRecord, targetCollector);
} else if (deserializationSchema instanceof
NativeKafkaConnectDeserializationSchema) {
((NativeKafkaConnectDeserializationSchema)
deserializationSchema)
.deserialize(consumerRecord, outputCollector);
@@ -89,6 +107,54 @@ public class KafkaRecordEmitter
splitState.setCurrentOffset(consumerRecord.offset() + 1);
}
+ private static Collector<SeaTunnelRow> headerInjectingCollector(
+ Collector<SeaTunnelRow> delegate, Headers headers, List<String>
headerFieldNames) {
+ return new Collector<SeaTunnelRow>() {
+ @Override
+ public void collect(SeaTunnelRow record) {
+ delegate.collect(appendHeaderFields(record, headers,
headerFieldNames));
+ }
+
+ @Override
+ public void collect(SchemaChangeEvent event) {
+ delegate.collect(event);
+ }
+
+ @Override
+ public void markSchemaChangeBeforeCheckpoint() {
+ delegate.markSchemaChangeBeforeCheckpoint();
+ }
+
+ @Override
+ public void markSchemaChangeAfterCheckpoint() {
+ delegate.markSchemaChangeAfterCheckpoint();
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return delegate.getCheckpointLock();
+ }
+ };
+ }
+
+ private static SeaTunnelRow appendHeaderFields(
+ SeaTunnelRow baseRow, Headers headers, List<String>
headerFieldNames) {
+ Object[] fields = new Object[baseRow.getFields().length +
headerFieldNames.size()];
+ System.arraycopy(baseRow.getFields(), 0, fields, 0,
baseRow.getFields().length);
+ for (int i = 0; i < headerFieldNames.size(); i++) {
+ Header header = headers.lastHeader(headerFieldNames.get(i));
+ fields[baseRow.getFields().length + i] =
+ header != null && header.value() != null
+ ? new String(header.value(),
StandardCharsets.UTF_8)
+ : null;
+ }
+ SeaTunnelRow newRow = new SeaTunnelRow(fields);
+ newRow.setRowKind(baseRow.getRowKind());
+ newRow.setTableId(baseRow.getTableId());
+ newRow.setOptions(baseRow.getOptions());
+ return newRow;
+ }
+
private static class OutputCollector<T> implements Collector<T> {
private Collector<T> output;
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
index bd689ecea4..63270eb58a 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java
@@ -35,6 +35,7 @@ import
org.apache.seatunnel.api.table.catalog.schema.ReadonlyConfigParser;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.MapType;
import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
@@ -172,11 +173,30 @@ public class KafkaSourceConfig implements Serializable {
consumerMetadata.setTopic(readonlyConfig.get(TOPIC));
consumerMetadata.setPattern(readonlyConfig.get(PATTERN));
consumerMetadata.setProperties(new Properties());
- // Create a catalog
- CatalogTable catalogTable = createCatalogTable(readonlyConfig);
- consumerMetadata.setCatalogTable(catalogTable);
- consumerMetadata.setDeserializationSchema(
- createDeserializationSchema(catalogTable, readonlyConfig));
+
+ CatalogTable baseCatalogTable = createCatalogTable(readonlyConfig);
+ List<String> headerFields =
+ readonlyConfig
+ .getOptional(KafkaSourceOptions.KAFKA_HEADERS_FIELDS)
+ .orElse(Collections.emptyList());
+ MessageFormat format = readonlyConfig.get(FORMAT);
+
+ boolean applyHeaderFields = !headerFields.isEmpty() && format !=
MessageFormat.NATIVE;
+
+ DeserializationSchema<SeaTunnelRow> schema =
+ createDeserializationSchema(baseCatalogTable, readonlyConfig,
headerFields);
+
+ CatalogTable outputCatalogTable =
+ applyHeaderFields
+ ? extendCatalogTableWithHeaderFields(baseCatalogTable,
headerFields)
+ : baseCatalogTable;
+
+ if (!headerFields.isEmpty() && format ==
MessageFormat.COMPATIBLE_KAFKA_CONNECT_JSON) {
+ consumerMetadata.setKafkaHeaderFields(headerFields);
+ }
+
+ consumerMetadata.setCatalogTable(outputCatalogTable);
+ consumerMetadata.setDeserializationSchema(schema);
// parse start mode
readonlyConfig
@@ -313,7 +333,7 @@ public class KafkaSourceConfig implements Serializable {
}
private DeserializationSchema<SeaTunnelRow> createDeserializationSchema(
- CatalogTable catalogTable, ReadonlyConfig readonlyConfig) {
+ CatalogTable catalogTable, ReadonlyConfig readonlyConfig,
List<String> headerFields) {
SeaTunnelRowType seaTunnelRowType = catalogTable.getSeaTunnelRowType();
MessageFormat format = readonlyConfig.get(FORMAT);
@@ -433,9 +453,57 @@ public class KafkaSourceConfig implements Serializable {
return schema;
}
+ if (!headerFields.isEmpty() && format != MessageFormat.NATIVE) {
+ SeaTunnelRowType baseRowType = (SeaTunnelRowType)
schema.getProducedType();
+ SeaTunnelRowType extendedRowType =
buildExtendedRowType(baseRowType, headerFields);
+ schema = new KafkaHeadersDeserializationSchema(schema,
headerFields, extendedRowType);
+ }
+
return new KafkaEventTimeDeserializationSchema(schema);
}
+ private SeaTunnelRowType buildExtendedRowType(
+ SeaTunnelRowType baseRowType, List<String> headerFieldNames) {
+ int baseCount = baseRowType.getTotalFields();
+ String[] fieldNames = new String[baseCount + headerFieldNames.size()];
+ SeaTunnelDataType<?>[] fieldTypes =
+ new SeaTunnelDataType[baseCount + headerFieldNames.size()];
+
+ System.arraycopy(baseRowType.getFieldNames(), 0, fieldNames, 0,
baseCount);
+ System.arraycopy(baseRowType.getFieldTypes(), 0, fieldTypes, 0,
baseCount);
+
+ for (int i = 0; i < headerFieldNames.size(); i++) {
+ fieldNames[baseCount + i] = headerFieldNames.get(i);
+ fieldTypes[baseCount + i] = BasicType.STRING_TYPE;
+ }
+
+ return new SeaTunnelRowType(fieldNames, fieldTypes);
+ }
+
+ private CatalogTable extendCatalogTableWithHeaderFields(
+ CatalogTable catalogTable, List<String> headerFieldNames) {
+ TableSchema baseSchema = catalogTable.getTableSchema();
+ TableSchema.Builder builder =
+ TableSchema.builder()
+ .columns(baseSchema.getColumns())
+ .primaryKey(baseSchema.getPrimaryKey())
+ .constraintKey(baseSchema.getConstraintKeys());
+
+ for (String headerField : headerFieldNames) {
+ builder.column(
+ PhysicalColumn.of(headerField, BasicType.STRING_TYPE, 0,
true, null, null));
+ }
+
+ return CatalogTable.of(
+ catalogTable.getTableId(),
+ builder.build(),
+ catalogTable.getOptions(),
+ catalogTable.getPartitionKeys(),
+ catalogTable.getComment(),
+ catalogTable.getCatalogName(),
+ catalogTable.getMetadataSchema());
+ }
+
private TableSchema nativeTableSchema() {
return TableSchema.builder()
.column(
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
index 6e54ea69fa..716163446e 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceFactory.java
@@ -65,7 +65,8 @@ public class KafkaSourceFactory implements TableSourceFactory
{
KafkaSourceOptions.DEBEZIUM_RECORD_INCLUDE_SCHEMA,
KafkaSourceOptions.DEBEZIUM_RECORD_TABLE_FILTER,
KafkaSourceOptions.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,
- KafkaSourceOptions.READER_CACHE_QUEUE_SIZE)
+ KafkaSourceOptions.READER_CACHE_QUEUE_SIZE,
+ KafkaSourceOptions.KAFKA_HEADERS_FIELDS)
.optional(
KafkaSourceOptions.START_MODE_TIMESTAMP,
Conditions.extension(
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaHeadersDeserializationSchemaTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaHeadersDeserializationSchemaTest.java
new file mode 100644
index 0000000000..f8a1260ae2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaHeadersDeserializationSchemaTest.java
@@ -0,0 +1,254 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.kafka.source;
+
+import org.apache.seatunnel.api.serialization.DeserializationSchema;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class KafkaHeadersDeserializationSchemaTest {
+
+ private static final SeaTunnelRowType BASE_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"user_id", "name"},
+ new SeaTunnelDataType[] {BasicType.INT_TYPE,
BasicType.STRING_TYPE});
+
+ private static final SeaTunnelRowType EXTENDED_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"user_id", "name", "correlation_id"},
+ new SeaTunnelDataType[] {
+ BasicType.INT_TYPE, BasicType.STRING_TYPE,
BasicType.STRING_TYPE
+ });
+
+ private static final List<String> HEADER_FIELDS =
Arrays.asList("correlation_id");
+
+ @Test
+ void testHeaderValueAppendedToRow() throws IOException {
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(BASE_ROW_TYPE, new Object[] {42,
"Alice"}),
+ HEADER_FIELDS,
+ EXTENDED_ROW_TYPE);
+
+ RecordHeaders headers = new RecordHeaders();
+ headers.add(
+ new RecordHeader("correlation_id",
"corr-001".getBytes(StandardCharsets.UTF_8)));
+ schema.setCurrentRecordHeaders(headers);
+
+ SeaTunnelRow row = schema.deserialize(new byte[0]);
+
+ Assertions.assertEquals(3, row.getFields().length);
+ Assertions.assertEquals(42, row.getField(0));
+ Assertions.assertEquals("Alice", row.getField(1));
+ Assertions.assertEquals("corr-001", row.getField(2));
+ }
+
+ @Test
+ void testNullHeaderValueBecomesNull() throws IOException {
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(BASE_ROW_TYPE, new Object[] {1,
"Bob"}),
+ HEADER_FIELDS,
+ EXTENDED_ROW_TYPE);
+
+ RecordHeaders headers = new RecordHeaders();
+ headers.add(new RecordHeader("correlation_id", null));
+ schema.setCurrentRecordHeaders(headers);
+
+ SeaTunnelRow row = schema.deserialize(new byte[0]);
+
+ Assertions.assertNull(row.getField(2));
+ }
+
+ @Test
+ void testMissingHeaderKeyBecomesNull() throws IOException {
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(BASE_ROW_TYPE, new Object[] {1,
"Charlie"}),
+ HEADER_FIELDS,
+ EXTENDED_ROW_TYPE);
+
+ schema.setCurrentRecordHeaders(new RecordHeaders());
+
+ SeaTunnelRow row = schema.deserialize(new byte[0]);
+
+ Assertions.assertNull(row.getField(2));
+ }
+
+ @Test
+ void testRowOptionsPreserved() throws IOException {
+ Map<String, Object> options = new HashMap<>();
+ options.put("EVENT_TIME", 12345L);
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(BASE_ROW_TYPE, new Object[] {1,
"Dave"}, options),
+ HEADER_FIELDS,
+ EXTENDED_ROW_TYPE);
+
+ RecordHeaders headers = new RecordHeaders();
+ headers.add(new RecordHeader("correlation_id",
"val".getBytes(StandardCharsets.UTF_8)));
+ schema.setCurrentRecordHeaders(headers);
+
+ SeaTunnelRow row = schema.deserialize(new byte[0]);
+
+ Assertions.assertEquals(12345L, row.getOptions().get("EVENT_TIME"));
+ }
+
+ @Test
+ void testGetProducedTypeReturnsExtended() {
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(BASE_ROW_TYPE, new Object[] {1,
"Eve"}),
+ HEADER_FIELDS,
+ EXTENDED_ROW_TYPE);
+
+ SeaTunnelRowType producedType = (SeaTunnelRowType)
schema.getProducedType();
+
+ Assertions.assertEquals(3, producedType.getTotalFields());
+ Assertions.assertEquals("user_id", producedType.getFieldName(0));
+ Assertions.assertEquals("name", producedType.getFieldName(1));
+ Assertions.assertEquals("correlation_id",
producedType.getFieldName(2));
+ Assertions.assertEquals(BasicType.STRING_TYPE,
producedType.getFieldType(2));
+ }
+
+ @Test
+ void testDeserializeViaCollector() throws IOException {
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(BASE_ROW_TYPE, new Object[] {5,
"Frank"}),
+ HEADER_FIELDS,
+ EXTENDED_ROW_TYPE);
+
+ RecordHeaders headers = new RecordHeaders();
+ headers.add(
+ new RecordHeader("correlation_id",
"corr-abc".getBytes(StandardCharsets.UTF_8)));
+ schema.setCurrentRecordHeaders(headers);
+
+ List<SeaTunnelRow> out = new ArrayList<>();
+ schema.deserialize(new byte[0], new TestCollector(out));
+
+ Assertions.assertEquals(1, out.size());
+ Assertions.assertEquals("corr-abc", out.get(0).getField(2));
+ }
+
+ @Test
+ void testMultipleHeaderFields() throws IOException {
+ SeaTunnelRowType multiExtendedType =
+ new SeaTunnelRowType(
+ new String[] {"user_id", "correlation_id", "trace_id"},
+ new SeaTunnelDataType[] {
+ BasicType.INT_TYPE, BasicType.STRING_TYPE,
BasicType.STRING_TYPE
+ });
+ List<String> multiHeaders = Arrays.asList("correlation_id",
"trace_id");
+
+ SeaTunnelRowType singleBase =
+ new SeaTunnelRowType(
+ new String[] {"user_id"}, new SeaTunnelDataType[]
{BasicType.INT_TYPE});
+
+ KafkaHeadersDeserializationSchema schema =
+ new KafkaHeadersDeserializationSchema(
+ new FixedRowSchema(singleBase, new Object[] {7}),
+ multiHeaders,
+ multiExtendedType);
+
+ RecordHeaders headers = new RecordHeaders();
+ headers.add(
+ new RecordHeader("correlation_id",
"corr-xyz".getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("trace_id",
"trace-123".getBytes(StandardCharsets.UTF_8)));
+ schema.setCurrentRecordHeaders(headers);
+
+ SeaTunnelRow row = schema.deserialize(new byte[0]);
+
+ Assertions.assertEquals(3, row.getFields().length);
+ Assertions.assertEquals(7, row.getField(0));
+ Assertions.assertEquals("corr-xyz", row.getField(1));
+ Assertions.assertEquals("trace-123", row.getField(2));
+ }
+
+ private static class FixedRowSchema implements
DeserializationSchema<SeaTunnelRow> {
+ private final SeaTunnelRowType rowType;
+ private final Object[] fields;
+ private final Map<String, Object> options;
+
+ FixedRowSchema(SeaTunnelRowType rowType, Object[] fields) {
+ this(rowType, fields, new HashMap<>());
+ }
+
+ FixedRowSchema(SeaTunnelRowType rowType, Object[] fields, Map<String,
Object> options) {
+ this.rowType = rowType;
+ this.fields = fields;
+ this.options = options;
+ }
+
+ @Override
+ public SeaTunnelRow deserialize(byte[] message) {
+ SeaTunnelRow row = new SeaTunnelRow(fields.clone());
+ row.setOptions(new HashMap<>(options));
+ return row;
+ }
+
+ @Override
+ public void deserialize(byte[] message, Collector<SeaTunnelRow> out)
throws IOException {
+ out.collect(deserialize(message));
+ }
+
+ @Override
+ public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
+ return rowType;
+ }
+ }
+
+ private static class TestCollector implements Collector<SeaTunnelRow> {
+ private final List<SeaTunnelRow> out;
+
+ TestCollector(List<SeaTunnelRow> out) {
+ this.out = out;
+ }
+
+ @Override
+ public void collect(SeaTunnelRow record) {
+ out.add(record);
+ }
+
+ @Override
+ public void collect(SchemaChangeEvent event) {}
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitterTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitterTest.java
index 14fd630304..9cc5ba3cfb 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitterTest.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitterTest.java
@@ -28,6 +28,8 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.kafka.config.MessageFormatErrorHandleWay;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -36,6 +38,7 @@ import org.mockito.Mockito;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -126,6 +129,108 @@ class KafkaRecordEmitterTest {
Assertions.assertEquals(6L, splitState.getCurrentOffset());
}
+ @Test
+ void emitRecordShouldAttachHeaderFields() throws Exception {
+ SeaTunnelRowType baseRowType =
+ new SeaTunnelRowType(
+ new String[] {"f0"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ SeaTunnelRowType extendedRowType =
+ new SeaTunnelRowType(
+ new String[] {"f0", "correlation_id"},
+ new SeaTunnelDataType[] {BasicType.STRING_TYPE,
BasicType.STRING_TYPE});
+ List<String> headerFields = Arrays.asList("correlation_id");
+
+ KafkaHeadersDeserializationSchema headersSchema =
+ new KafkaHeadersDeserializationSchema(
+ new SimpleStringRowSchema(baseRowType), headerFields,
extendedRowType);
+ DeserializationSchema<SeaTunnelRow> schema =
+ new KafkaEventTimeDeserializationSchema(headersSchema);
+
+ ConsumerMetadata metadata = new ConsumerMetadata();
+ metadata.setDeserializationSchema(schema);
+ Map<TablePath, ConsumerMetadata> map = new HashMap<>();
+ TablePath tablePath = TablePath.DEFAULT;
+ map.put(tablePath, metadata);
+
+ KafkaRecordEmitter emitter = new KafkaRecordEmitter(map,
MessageFormatErrorHandleWay.FAIL);
+
+ RecordHeaders recordHeaders = new RecordHeaders();
+ recordHeaders.add(
+ new RecordHeader("correlation_id",
"corr-test".getBytes(StandardCharsets.UTF_8)));
+
+ org.apache.kafka.clients.consumer.ConsumerRecord<byte[], byte[]>
record =
+
Mockito.mock(org.apache.kafka.clients.consumer.ConsumerRecord.class);
+ Mockito.when(record.timestamp()).thenReturn(1000L);
+
Mockito.when(record.value()).thenReturn("hello".getBytes(StandardCharsets.UTF_8));
+ Mockito.when(record.offset()).thenReturn(0L);
+ Mockito.when(record.headers()).thenReturn(recordHeaders);
+
+ KafkaSourceSplit split = new KafkaSourceSplit(tablePath, new
TopicPartition("t", 0));
+ KafkaSourceSplitState splitState = new KafkaSourceSplitState(split);
+
+ List<SeaTunnelRow> out = new ArrayList<>();
+ emitter.emitRecord(record, new TestCollector(out), splitState);
+
+ Assertions.assertEquals(1, out.size());
+ SeaTunnelRow row = out.get(0);
+ Assertions.assertEquals(2, row.getFields().length);
+ Assertions.assertEquals("hello", row.getField(0));
+ Assertions.assertEquals("corr-test", row.getField(1));
+ }
+
+ @Test
+ void emitRecordShouldPreserveEventTimeWithHeaders() throws Exception {
+ long kafkaTimestamp = 9876543210L;
+ SeaTunnelRowType baseRowType =
+ new SeaTunnelRowType(
+ new String[] {"f0"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+ SeaTunnelRowType extendedRowType =
+ new SeaTunnelRowType(
+ new String[] {"f0", "trace_id"},
+ new SeaTunnelDataType[] {BasicType.STRING_TYPE,
BasicType.STRING_TYPE});
+
+ KafkaHeadersDeserializationSchema headersSchema =
+ new KafkaHeadersDeserializationSchema(
+ new SimpleStringRowSchema(baseRowType),
+ Arrays.asList("trace_id"),
+ extendedRowType);
+ DeserializationSchema<SeaTunnelRow> schema =
+ new KafkaEventTimeDeserializationSchema(headersSchema);
+
+ ConsumerMetadata metadata = new ConsumerMetadata();
+ metadata.setDeserializationSchema(schema);
+ Map<TablePath, ConsumerMetadata> map = new HashMap<>();
+ TablePath tablePath = TablePath.DEFAULT;
+ map.put(tablePath, metadata);
+
+ KafkaRecordEmitter emitter = new KafkaRecordEmitter(map,
MessageFormatErrorHandleWay.FAIL);
+
+ RecordHeaders recordHeaders = new RecordHeaders();
+ recordHeaders.add(
+ new RecordHeader("trace_id",
"trace-abc".getBytes(StandardCharsets.UTF_8)));
+
+ org.apache.kafka.clients.consumer.ConsumerRecord<byte[], byte[]>
record =
+
Mockito.mock(org.apache.kafka.clients.consumer.ConsumerRecord.class);
+ Mockito.when(record.timestamp()).thenReturn(kafkaTimestamp);
+
Mockito.when(record.value()).thenReturn("data".getBytes(StandardCharsets.UTF_8));
+ Mockito.when(record.offset()).thenReturn(0L);
+ Mockito.when(record.headers()).thenReturn(recordHeaders);
+
+ KafkaSourceSplit split = new KafkaSourceSplit(tablePath, new
TopicPartition("t3", 0));
+ KafkaSourceSplitState splitState = new KafkaSourceSplitState(split);
+
+ List<SeaTunnelRow> out = new ArrayList<>();
+ emitter.emitRecord(record, new TestCollector(out), splitState);
+
+ Assertions.assertEquals(1, out.size());
+ SeaTunnelRow row = out.get(0);
+ // Header field populated
+ Assertions.assertEquals("trace-abc", row.getField(1));
+ // Event time also set
+ Assertions.assertEquals(
+ kafkaTimestamp,
row.getOptions().get(CommonOptions.EVENT_TIME.getName()));
+ }
+
private static class SimpleStringRowSchema implements
DeserializationSchema<SeaTunnelRow> {
private final SeaTunnelRowType producedType;
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
index 2a3b63a944..e4dd9dc6eb 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java
@@ -19,14 +19,19 @@ package
org.apache.seatunnel.connectors.seatunnel.kafka.source;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.format.json.debezium.DebeziumJsonDeserializationSchemaDispatcher;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static
org.apache.seatunnel.api.options.ConnectorCommonOptions.DATABASE_NAME;
@@ -78,6 +83,117 @@ public class KafkaSourceConfigTest {
.get(TablePath.of("test.test.test")));
}
+ @Test
+ void testKafkaHeaderFieldsExtendsOutputSchema() {
+ Map<String, Object> schemaFields = new HashMap<>();
+ schemaFields.put("user_id", "int");
+ schemaFields.put("name", "string");
+
+ Map<String, Object> schema = new HashMap<>();
+ schema.put("fields", schemaFields);
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("bootstrap.servers", "localhost:9092");
+ configMap.put("topic", "test");
+ configMap.put("schema", schema);
+ configMap.put("format", "json");
+ configMap.put("kafka_headers_fields", Arrays.asList("correlation_id",
"trace_id"));
+
+ KafkaSourceConfig sourceConfig = new
KafkaSourceConfig(ReadonlyConfig.fromMap(configMap));
+
+ CatalogTable catalogTable =
+
sourceConfig.getMapMetadata().get(TablePath.of("test")).getCatalogTable();
+ SeaTunnelRowType rowType = catalogTable.getSeaTunnelRowType();
+
+ // Schema should be extended with the 2 header fields
+ Assertions.assertEquals(4, rowType.getTotalFields());
+ List<String> fieldNames = Arrays.asList(rowType.getFieldNames());
+ Assertions.assertTrue(fieldNames.contains("user_id"));
+ Assertions.assertTrue(fieldNames.contains("name"));
+ Assertions.assertTrue(fieldNames.contains("correlation_id"));
+ Assertions.assertTrue(fieldNames.contains("trace_id"));
+ // Header fields are STRING type
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
rowType.getFieldType(rowType.indexOf("correlation_id")));
+ Assertions.assertEquals(
+ BasicType.STRING_TYPE,
rowType.getFieldType(rowType.indexOf("trace_id")));
+ }
+
+ @Test
+ void testKafkaHeaderFieldsDeserializationSchemaWrapped() {
+ Map<String, Object> schemaFields = new HashMap<>();
+ schemaFields.put("user_id", "int");
+
+ Map<String, Object> schema = new HashMap<>();
+ schema.put("fields", schemaFields);
+
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("bootstrap.servers", "localhost:9092");
+ configMap.put("topic", "test");
+ configMap.put("schema", schema);
+ configMap.put("format", "json");
+ configMap.put("kafka_headers_fields", Arrays.asList("correlation_id"));
+
+ KafkaSourceConfig sourceConfig = new
KafkaSourceConfig(ReadonlyConfig.fromMap(configMap));
+
+ DeserializationSchema<SeaTunnelRow> deserializationSchema =
+
sourceConfig.getMapMetadata().get(TablePath.of("test")).getDeserializationSchema();
+
+ // Outer wrapper: KafkaEventTimeDeserializationSchema
+ Assertions.assertInstanceOf(
+ KafkaEventTimeDeserializationSchema.class,
deserializationSchema);
+ DeserializationSchema<SeaTunnelRow> inner =
+ ((KafkaEventTimeDeserializationSchema)
deserializationSchema).getDelegate();
+
+ // Inner wrapper: KafkaHeadersDeserializationSchema
+ Assertions.assertInstanceOf(KafkaHeadersDeserializationSchema.class,
inner);
+ SeaTunnelRowType producedType = (SeaTunnelRowType)
inner.getProducedType();
+ Assertions.assertEquals(2, producedType.getTotalFields());
+ Assertions.assertEquals("correlation_id",
producedType.getFieldName(1));
+ }
+
+ @Test
+ void
testKafkaHeaderFieldsExtendsSchemaForCompatibleKafkaConnectJsonFormat() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("bootstrap.servers", "localhost:9092");
+ configMap.put("topic", "test");
+ configMap.put("format", "compatible_kafka_connect_json");
+ configMap.put("kafka_headers_fields", Arrays.asList("correlation_id"));
+
+ KafkaSourceConfig sourceConfig = new
KafkaSourceConfig(ReadonlyConfig.fromMap(configMap));
+
+ ConsumerMetadata metadata =
sourceConfig.getMapMetadata().get(TablePath.of("test"));
+
+ // Header fields stored for emitter-side injection
+ Assertions.assertEquals(Arrays.asList("correlation_id"),
metadata.getKafkaHeaderFields());
+
+ // Output schema includes header field
+ SeaTunnelRowType rowType =
metadata.getCatalogTable().getSeaTunnelRowType();
+ List<String> fieldNames = Arrays.asList(rowType.getFieldNames());
+ Assertions.assertTrue(fieldNames.contains("correlation_id"));
+ }
+
+ @Test
+ void testKafkaHeaderFieldsNotAppliedToNativeFormat() {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("bootstrap.servers", "localhost:9092");
+ configMap.put("topic", "test");
+ configMap.put("format", "native");
+ configMap.put("kafka_headers_fields", Arrays.asList("correlation_id"));
+
+ KafkaSourceConfig sourceConfig = new
KafkaSourceConfig(ReadonlyConfig.fromMap(configMap));
+
+ CatalogTable catalogTable =
+
sourceConfig.getMapMetadata().get(TablePath.of("test")).getCatalogTable();
+ SeaTunnelRowType rowType = catalogTable.getSeaTunnelRowType();
+
+ // NATIVE format already has its own fixed schema; header fields
should NOT be added
+ List<String> fieldNames = Arrays.asList(rowType.getFieldNames());
+ Assertions.assertFalse(
+ fieldNames.contains("correlation_id"),
+ "NATIVE format should not add kafka_headers_fields to schema");
+ }
+
@Test
void testDeserializationWithSchema() {
Map<String, Object> schemaFields = new HashMap<>();
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index cccefc0455..f906bab4c6 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -394,6 +394,64 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
}
+ @TestTemplate
+ public void testSourceKafkaWithHeaders(TestContainer container)
+ throws IOException, InterruptedException {
+ String topicName = "test_topic_source_headers";
+
+ // Produce 10 records with JSON value and Kafka headers
+ try {
+ for (int i = 0; i < 10; i++) {
+ byte[] value =
+ ("{\"user_id\":" + i + ",\"name\":\"user_" + i + "\"}")
+ .getBytes(StandardCharsets.UTF_8);
+ Header corrHeader =
+ new RecordHeader(
+ "correlation_id", ("corr-" +
i).getBytes(StandardCharsets.UTF_8));
+ Header traceHeader =
+ new RecordHeader(
+ "trace_id", ("trace-" +
i).getBytes(StandardCharsets.UTF_8));
+ ProducerRecord<byte[], byte[]> record =
+ new ProducerRecord<>(
+ topicName,
+ null,
+ null,
+ null,
+ value,
+ Arrays.asList(corrHeader, traceHeader));
+ producer.send(record).get();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ producer.flush();
+
+ Container.ExecResult execResult =
+ container.executeJob("/kafka/kafka_source_with_headers.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+
+ // Additional verification: read records back and check header fields
are accessible
+ List<ConsumerRecord<String, String>> records =
getKafkaRecordData(topicName);
+ Assertions.assertEquals(10, records.size());
+
+ for (int i = 0; i < records.size(); i++) {
+ ConsumerRecord<String, String> record = records.get(i);
+ Map<String, String> headers =
convertHeadersToMap(record.headers());
+
+ Assertions.assertTrue(
+ headers.containsKey("correlation_id"),
+ "Record should contain 'correlation_id' header");
+ Assertions.assertTrue(
+ headers.containsKey("trace_id"), "Record should contain
'trace_id' header");
+ Assertions.assertTrue(
+ headers.get("correlation_id").startsWith("corr-"),
+ "correlation_id header value should start with 'corr-'");
+ Assertions.assertTrue(
+ headers.get("trace_id").startsWith("trace-"),
+ "trace_id header value should start with 'trace-'");
+ }
+ }
+
@TestTemplate
public void testDefaultRandomSinkKafka(TestContainer container)
throws IOException, InterruptedException {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_source_with_headers.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_source_with_headers.conf
new file mode 100644
index 0000000000..587c0e11a2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_source_with_headers.conf
@@ -0,0 +1,94 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 1
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+}
+
+source {
+ Kafka {
+ bootstrap.servers = "kafkaCluster:9092"
+ topic = "test_topic_source_headers"
+ start_mode = "earliest"
+ kafka_headers_fields = ["correlation_id", "trace_id"]
+ schema = {
+ fields {
+ user_id = "int"
+ name = "string"
+ }
+ }
+ format = json
+ }
+}
+
+sink {
+ Assert {
+ rules = {
+ field_rules = [
+ {
+ field_name = user_id
+ field_type = int
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ }
+ ]
+ },
+ {
+ field_name = name
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ }
+ ]
+ },
+ {
+ field_name = correlation_id
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ }
+ ]
+ },
+ {
+ field_name = trace_id
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ }
+ ]
+ }
+ ]
+ row_rules = [
+ {
+ rule_type = MIN_ROW
+ rule_value = 10
+ }
+ ]
+ }
+ }
+}
