This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12967-41679ddc46117c09f42a2c9229e9148f1a56d065 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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; + } + } +}
