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 acfdbd4637 [Fix][Connector-V2] Support Kafka Avro writer schema
(#11303)
acfdbd4637 is described below
commit acfdbd46377e569dfe6ebb9cc3e39b6aa25b24e5
Author: Jast <[email protected]>
AuthorDate: Sat Jul 4 13:59:21 2026 +0800
[Fix][Connector-V2] Support Kafka Avro writer schema (#11303)
---
docs/en/connectors/source/Kafka.md | 1 +
docs/zh/connectors/source/Kafka.md | 1 +
.../seatunnel/kafka/config/KafkaSourceOptions.java | 9 ++++
.../seatunnel/kafka/source/KafkaSourceConfig.java | 10 ++++-
.../seatunnel/kafka/source/KafkaSourceFactory.java | 1 +
.../format/avro/AvroDeserializationSchema.java | 6 +++
.../seatunnel/format/avro/AvroToRowConverter.java | 7 +++
.../format/avro/AvroSerializationSchemaTest.java | 50 ++++++++++++++++++++++
8 files changed, 84 insertions(+), 1 deletion(-)
diff --git a/docs/en/connectors/source/Kafka.md
b/docs/en/connectors/source/Kafka.md
index c68dadf137..95fb278fdb 100644
--- a/docs/en/connectors/source/Kafka.md
+++ b/docs/en/connectors/source/Kafka.md
@@ -47,6 +47,7 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| kafka.config | Map
| No | - | In
addition to the above necessary parameters that must be specified by the `Kafka
consumer` client, users can also specify multiple `consumer` client
non-mandatory parameters, covering [all consumer parameters specified in the
official Kafka
document](https://kafka.apache.org/documentation.html#consumerconfigs).
[...]
| schema | Config
| No | - | The
structure of the data, including field names and field types. For more details,
please refer to [Schema
Feature](../../introduction/concepts/schema-feature.md).
[...]
| format | String
| No | json |
Data format. The default format is json. Optional text format, canal_json,
debezium_json, maxwell_json, ogg_json, avro , protobuf and native. If you use
json or text format. The default field separator is ", ". If you customize the
delimiter, add the "field_delimiter" option.If you use canal format, please
refer to [canal-json](../form [...]
+| avro_schema | String
| No | - |
Effective when `format` is `avro`. Provides the writer Avro schema used to
deserialize binary Avro messages whose record name, namespace, or union layout
differs from the SeaTunnel schema.
[...]
| format_error_handle_way | String
| No | fail | The
processing method of data format error. The default value is fail, and the
optional value is (fail, skip). When fail is selected, data format error will
block and an exception will be thrown. When skip is selected, data format error
will skip this line data.
[...]
| debezium_record_table_filter | Config
| No | - |
Used for filtering data in debezium format, only when the format is set to
`debezium_json`. Please refer `debezium_record_table_filter` below
[...]
| field_delimiter | String
| No | , |
Customize the field delimiter for data format.
[...]
diff --git a/docs/zh/connectors/source/Kafka.md
b/docs/zh/connectors/source/Kafka.md
index 0b396000b0..bc35e3ab9f 100644
--- a/docs/zh/connectors/source/Kafka.md
+++ b/docs/zh/connectors/source/Kafka.md
@@ -47,6 +47,7 @@ import ChangeLog from '../changelog/connector-kafka.md';
| kafka.config | Map |
否 | - | 除了上述必要参数外，用户还可以指定多个非强制的消费者客户端参数，覆盖 [Kafka
官方文档](https://kafka.apache.org/documentation.html#consumerconfigs) 中指定的所有消费者参数。
|
| schema | Config |
否 | - | 数据结构，包括字段名称和字段类型。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
| format | String |
否 | json | 数据格式。默认格式为 json。可选格式包括 text, canal_json,
debezium_json, ogg_json, maxwell_json, avro , protobuf和native。默认字段分隔符为 ",
"。如果自定义分隔符，添加 "field_delimiter" 选项。如果使用 canal 格式，请参考
[canal-json](../formats/canal-json.md) 了解详细信息。如果使用 debezium 格式，请参考
[debezium-json](../formats/debezium-json.md)。一些Format的详细信息请参考
[formats](../formats) |
+| avro_schema | String |
否 | - | 当 `format` 为 `avro` 时生效。用于提供二进制 Avro 消息的
writer schema，适用于消息的 record 名称、namespace 或 union 结构与 SeaTunnel schema 不完全一致的场景。
|
| format_error_handle_way | String |
否 | fail | 数据格式错误的处理方式。默认值为 fail，可选值为 fail 和
skip。当选择 fail 时，数据格式错误将阻塞并抛出异常。当选择 skip 时，数据格式错误将跳过此行数据。
|
| debezium_record_table_filter | Config |
否 | - | 用于过滤 debezium 格式的数据，仅当格式设置为
`debezium_json` 时使用。请参阅下面的 `debezium_record_table_filter`
|
| field_delimiter | String |
否 | , | 自定义数据格式的字段分隔符。
|
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 e351c7e26b..9373290978 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
@@ -137,4 +137,13 @@ public class KafkaSourceOptions extends KafkaBaseOptions {
"Whether to strip the Confluent Schema Registry
wire format header "
+ "(magic byte, schema id and message
indexes) before "
+ "protobuf deserialization.");
+
+ public static final Option<String> AVRO_SCHEMA =
+ Options.key("avro_schema")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "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.");
}
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 249030c736..0913f47ebf 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
@@ -81,6 +81,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaBaseCo
import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaBaseConstants.TIMESTAMP;
import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaBaseConstants.TIMESTAMP_TYPE;
import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaBaseConstants.VALUE;
+import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSourceOptions.AVRO_SCHEMA;
import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSourceOptions.BOOTSTRAP_SERVERS;
import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSourceOptions.COMMIT_ON_CHECKPOINT;
import static
org.apache.seatunnel.connectors.seatunnel.kafka.config.KafkaSourceOptions.CONSUMER_GROUP;
@@ -414,7 +415,14 @@ public class KafkaSourceConfig implements Serializable {
}
break;
case AVRO:
- schema = new AvroDeserializationSchema(catalogTable);
+ Optional<String> avroSchema =
readonlyConfig.getOptional(AVRO_SCHEMA);
+ schema =
+ avroSchema
+ .map(
+ writerSchema ->
+ new
AvroDeserializationSchema(
+ catalogTable,
writerSchema))
+ .orElseGet(() -> new
AvroDeserializationSchema(catalogTable));
break;
case PROTOBUF:
boolean stripSchemaRegistryHeader =
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 1295cd68d2..4751d7982a 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
@@ -56,6 +56,7 @@ public class KafkaSourceFactory implements TableSourceFactory
{
KafkaSourceOptions.KAFKA_CONFIG,
KafkaSourceOptions.SCHEMA,
KafkaSourceOptions.FORMAT,
+ KafkaSourceOptions.AVRO_SCHEMA,
KafkaSourceOptions.DEBEZIUM_RECORD_INCLUDE_SCHEMA,
KafkaSourceOptions.DEBEZIUM_RECORD_TABLE_FILTER,
KafkaSourceOptions.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
index 32396c245e..c928bbb6be 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroDeserializationSchema.java
@@ -45,6 +45,12 @@ public class AvroDeserializationSchema implements
DeserializationSchema<SeaTunne
this.converter = new AvroToRowConverter(rowType);
}
+ public AvroDeserializationSchema(CatalogTable catalogTable, String
writerSchema) {
+ this.catalogTable = catalogTable;
+ this.rowType = catalogTable.getSeaTunnelRowType();
+ this.converter = new AvroToRowConverter(rowType, writerSchema);
+ }
+
@Override
public SeaTunnelRow deserialize(byte[] message) throws IOException {
BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(message,
null);
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
index 84b1063600..af7912d702 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java
@@ -51,6 +51,13 @@ public class AvroToRowConverter implements Serializable {
schema =
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(rowType);
}
+ public AvroToRowConverter(SeaTunnelRowType rowType, String writerSchema) {
+ schema =
+ writerSchema == null
+ ?
SeaTunnelRowTypeToAvroSchemaConverter.buildAvroSchemaWithRowType(rowType)
+ : new Schema.Parser().parse(writerSchema);
+ }
+
public DatumReader<GenericRecord> getReader() {
if (reader == null) {
reader = createReader();
diff --git
a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
index 18ca6d06e0..10eb85079f 100644
---
a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
+++
b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroSerializationSchemaTest.java
@@ -28,9 +28,17 @@ 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.avro.Schema;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDate;
@@ -235,4 +243,46 @@ class AvroSerializationSchemaTest {
Assertions.assertEquals(subRow.getField(12), null);
Assertions.assertEquals(subRow.getField(13), null);
}
+
+ @Test
+ public void testDeserializeWithProvidedWriterSchema() throws IOException {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"dev_id", "timestamp", "s_port"},
+ new SeaTunnelDataType<?>[] {
+ BasicType.STRING_TYPE, BasicType.LONG_TYPE,
BasicType.INT_TYPE
+ });
+ CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "",
"", "test", rowType);
+ String writerSchemaText =
+ "{"
+ + "\"type\":\"record\","
+ + "\"name\":\"AvroEvent\","
+ + "\"namespace\":\"safe.serialize\","
+ + "\"fields\":["
+ +
"{\"name\":\"dev_id\",\"type\":[{\"type\":\"string\",\"avro.java.string\":\"String\"},\"null\"]},"
+ +
"{\"name\":\"timestamp\",\"type\":[\"long\",\"null\"]},"
+ + "{\"name\":\"s_port\",\"type\":[\"int\",\"null\"]}"
+ + "]"
+ + "}";
+ Schema writerSchema = new Schema.Parser().parse(writerSchemaText);
+ GenericRecord record =
+ new GenericRecordBuilder(writerSchema)
+ .set("dev_id", "device-1")
+ .set("timestamp", 1769504419000L)
+ .set("s_port", null)
+ .build();
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(out,
null);
+ new GenericDatumWriter<GenericRecord>(writerSchema).write(record,
encoder);
+ encoder.flush();
+
+ AvroDeserializationSchema deserializationSchema =
+ new AvroDeserializationSchema(catalogTable, writerSchemaText);
+ SeaTunnelRow row =
deserializationSchema.deserialize(out.toByteArray());
+
+ Assertions.assertEquals("device-1", row.getField(0));
+ Assertions.assertEquals(1769504419000L, row.getField(1));
+ Assertions.assertNull(row.getField(2));
+ }
}
