This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 76e11cb80e [Fix][Format] Produce no row for a kafka connect json
tombstone (#12967)
76e11cb80e is described below
commit 76e11cb80e36b4844ba6820d43e0522dbd912d3c
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Tue Oct 27 03:49:13 2026 +0000
[Fix][Format] Produce no row for a kafka connect json tombstone (#12967)
---
.../formats/kafka-compatible-kafkaconnect-json.md | 4 +
docs/en/connectors/source/Kafka.md | 11 +-
.../formats/kafka-compatible-kafkaconnect-json.md | 3 +
docs/zh/connectors/source/Kafka.md | 7 +-
...ompatibleKafkaConnectDeserializationSchema.java | 7 +
...tibleKafkaConnectDeserializationSchemaTest.java | 141 +++++++++++++++++++++
6 files changed, 165 insertions(+), 8 deletions(-)
diff --git a/docs/en/connectors/formats/kafka-compatible-kafkaconnect-json.md
b/docs/en/connectors/formats/kafka-compatible-kafkaconnect-json.md
index 97c4052d88..6054b42e95 100644
--- a/docs/en/connectors/formats/kafka-compatible-kafkaconnect-json.md
+++ b/docs/en/connectors/formats/kafka-compatible-kafkaconnect-json.md
@@ -6,6 +6,10 @@ Seatunnel connector kafka supports parsing data extracted
through kafka connect
UPDATE messages must contain a complete row image; partial updates containing
only changed fields are not supported.
+A tombstone, a record that carries a key and a `null` value, produces no row.
The source passes it over and the
+offset advances, so it is not treated as a parse error. A record whose value
is present but zero length is a
+different case and is still a decode error.
+
# How To Use
## Kafka Sink Mysql
diff --git a/docs/en/connectors/source/Kafka.md
b/docs/en/connectors/source/Kafka.md
index 4d4e1183a2..a42a18ce82 100644
--- a/docs/en/connectors/source/Kafka.md
+++ b/docs/en/connectors/source/Kafka.md
@@ -617,11 +617,12 @@ Note：key/value is of type byte[].
### Tombstone records
A tombstone is a Kafka record that has a key and no value, which compacted
topics use to mark a
-key as deleted. For the `json`, `text`, `canal_json`, `debezium_json`,
`ogg_json`, `maxwell_json`
-and `avro` formats the source reads a tombstone as a record that produces no
row: it is not a
-format error, so it is passed over whatever `format_error_handle_way` is set
to, and the offset
-moves on past it. The `text`, `canal_json`, `debezium_json`, `ogg_json` and
`avro` formats treat
-an empty (zero length) value the same way; `json` and `maxwell_json` pass over
a null value only.
+key as deleted. For the `json`, `text`, `canal_json`, `debezium_json`,
`ogg_json`, `maxwell_json`,
+`avro` and `compatible_kafka_connect_json` formats the source reads a
tombstone as a record that
+produces no row: it is not a format error, so it is passed over whatever
`format_error_handle_way`
+is set to, and the offset moves on past it. The `text`, `canal_json`,
`debezium_json`, `ogg_json`
+and `avro` formats treat an empty (zero length) value the same way; `json`,
`maxwell_json` and
+`compatible_kafka_connect_json` pass over a null value only.
The `protobuf` format has no such guard and is not covered by this behaviour.
diff --git a/docs/zh/connectors/formats/kafka-compatible-kafkaconnect-json.md
b/docs/zh/connectors/formats/kafka-compatible-kafkaconnect-json.md
index a50f8039c3..421c3a149c 100644
--- a/docs/zh/connectors/formats/kafka-compatible-kafkaconnect-json.md
+++ b/docs/zh/connectors/formats/kafka-compatible-kafkaconnect-json.md
@@ -6,6 +6,9 @@ Seatunnel 的 Kafka 连接器支持解析通过 Kafka Connect Source 抽取的
更新消息必须包含完整行数据；仅包含变更字段的部分更新暂不支持。
+墓碑消息（tombstone，即带有键且值为 `null` 的记录）不会产生任何行。Source 会跳过该记录并推进 offset，因此它不会被视为解析错误。
+值存在但长度为 0 的记录属于另一种情况，仍然是解码错误。
+
# 如何使用
## Kafka 流入 Mysql
diff --git a/docs/zh/connectors/source/Kafka.md
b/docs/zh/connectors/source/Kafka.md
index baf14efa39..e238206d2e 100644
--- a/docs/zh/connectors/source/Kafka.md
+++ b/docs/zh/connectors/source/Kafka.md
@@ -607,9 +607,10 @@ source {
### 墓碑消息（tombstone）
墓碑消息是只有 key、没有 value 的 Kafka 记录，压缩主题用它标记某个 key 已被删除。对于
`json`、`text`、`canal_json`、
-`debezium_json`、`ogg_json`、`maxwell_json` 和 `avro`
格式，源端把墓碑消息读取为不产生任何数据行的记录：它不属于格式错误，因此无论
-`format_error_handle_way` 取何值都会被跳过，并且位点会继续向前推进。其中
`text`、`canal_json`、`debezium_json`、`ogg_json`
-和 `avro` 对空（长度为 0）的 value 也做同样处理；`json` 和 `maxwell_json` 仅跳过 null value。
+`debezium_json`、`ogg_json`、`maxwell_json`、`avro` 和
`compatible_kafka_connect_json` 格式，源端把墓碑消息读取为不产生
+任何数据行的记录：它不属于格式错误，因此无论 `format_error_handle_way` 取何值都会被跳过，并且位点会继续向前推进。其中
`text`、
+`canal_json`、`debezium_json`、`ogg_json` 和 `avro` 对空（长度为 0）的 value
也做同样处理；`json`、`maxwell_json` 和
+`compatible_kafka_connect_json` 仅跳过 null value。
`protobuf` 格式没有这样的空值判断，不适用上述行为。
diff --git
a/seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java
b/seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java
index 0a9e8867ea..1b55d09f46 100644
---
a/seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java
+++
b/seatunnel-formats/seatunnel-format-compatible-connect-json/src/main/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java
@@ -110,6 +110,13 @@ public class CompatibleKafkaConnectDeserializationSchema
if (msg == null) {
return;
}
+ if (msg.value() == null) {
+ // A Kafka tombstone carries a null value to mark its key deleted.
That is a
+ // deletion marker rather than a malformed message, so it produces
no row and the
+ // source moves past it, as the json and maxwell_json formats
already do. A zero
+ // length payload is deliberately not covered: it is a decode
error here, as before.
+ return;
+ }
SinkRecord record = convertToSinkRecord(msg);
RowKind rowKind = RowKind.INSERT;
JsonNode jsonNode =
diff --git
a/seatunnel-formats/seatunnel-format-compatible-connect-json/src/test/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchemaTest.java
b/seatunnel-formats/seatunnel-format-compatible-connect-json/src/test/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchemaTest.java
new file mode 100644
index 0000000000..db9c3de7f5
--- /dev/null
+++
b/seatunnel-formats/seatunnel-format-compatible-connect-json/src/test/java/org/apache/seatunnel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchemaTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.format.compatible.kafka.connect.json;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+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.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.TimestampType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** Tombstone handling for the {@code compatible_kafka_connect_json} source
format. */
+class CompatibleKafkaConnectDeserializationSchemaTest {
+
+ /**
+ * The converters are configured with {@code schemas.enable}, whose option
default is true, so
+ * both the key and the value of a well formed record carry a
schema/payload envelope.
+ */
+ private static final String KEY_ENVELOPE =
+
"{\"schema\":{\"type\":\"string\",\"optional\":true},\"payload\":\"k1\"}";
+
+ private static final String VALUE_ENVELOPE =
+ "{\"schema\":{\"type\":\"struct\",\"fields\":["
+ +
"{\"type\":\"int64\",\"optional\":true,\"field\":\"id\"},"
+ +
"{\"type\":\"string\",\"optional\":true,\"field\":\"name\"}]},"
+ + "\"payload\":{\"id\":1,\"name\":\"a\"}}";
+
+ /** A tombstone carries a null value, so it must produce no row rather
than failing. */
+ @Test
+ void deserializeProducesNoRowForATombstone() throws Exception {
+ RecordingCollector collector = new RecordingCollector();
+ schema().deserialize(record(null), collector);
+
+ Assertions.assertTrue(
+ collector.rows.isEmpty(), "a tombstone must not produce a row
of its own");
+ }
+
+ /** The guard must not swallow ordinary records. */
+ @Test
+ void deserializeStillReadsAWellFormedRecord() throws Exception {
+ RecordingCollector collector = new RecordingCollector();
+
schema().deserialize(record(VALUE_ENVELOPE.getBytes(StandardCharsets.UTF_8)),
collector);
+
+ Assertions.assertEquals(1, collector.rows.size(), "a valid record must
still be read");
+ SeaTunnelRow row = collector.rows.get(0);
+ Assertions.assertEquals(1L, row.getField(0));
+ Assertions.assertEquals("a", row.getField(1));
+ }
+
+ /**
+ * A zero length value is a decode error rather than a deletion marker, so
the guard must not
+ * pass it over. Asserted against the contract rather than against a
specific exception type,
+ * which comes from the Kafka converter.
+ */
+ @Test
+ void deserializeDoesNotTreatAnEmptyValueAsATombstone() {
+ RecordingCollector collector = new RecordingCollector();
+
+ Assertions.assertThrows(
+ Exception.class,
+ () -> schema().deserialize(record(new byte[0]), collector),
+ "a zero length value must stay a decode error, not be read as
a tombstone");
+ Assertions.assertTrue(
+ collector.rows.isEmpty(), "a rejected record must not have
produced a row");
+ }
+
+ private static CompatibleKafkaConnectDeserializationSchema schema() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "name"},
+ new SeaTunnelDataType[] {BasicType.LONG_TYPE,
BasicType.STRING_TYPE});
+ CatalogTable catalogTable =
CatalogTableUtil.getCatalogTable("kafka_connect_json", rowType);
+ // The same flags KafkaSourceConfig passes, both of which default to
true.
+ return new CompatibleKafkaConnectDeserializationSchema(
+ catalogTable, true, true, false, false);
+ }
+
+ private static ConsumerRecord<byte[], byte[]> record(byte[] value) {
+ return new ConsumerRecord<>(
+ "topic",
+ 0,
+ 42L,
+ 1234L,
+ TimestampType.CREATE_TIME,
+ 0,
+ 0,
+ KEY_ENVELOPE.getBytes(StandardCharsets.UTF_8),
+ value,
+ new RecordHeaders(),
+ Optional.empty());
+ }
+
+ private static class RecordingCollector implements Collector<SeaTunnelRow>
{
+ private final List<SeaTunnelRow> rows = new ArrayList<>();
+
+ @Override
+ public void collect(SeaTunnelRow record) {
+ rows.add(record);
+ }
+
+ @Override
+ public void collect(SchemaChangeEvent event) {}
+
+ @Override
+ public void markSchemaChangeBeforeCheckpoint() {}
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+ }
+}
