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-12953-eee6f703f5e689f17127f65b3eb0bdcb73b6180e in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 32e2efb2f89351049e4ddca4c494d5d476fbe5cb Author: Sepuri Sai Krishna <[email protected]> AuthorDate: Wed Oct 28 16:55:27 2026 +0000 [Fix][Format][Protobuf] Produce no row for a tombstone instead of failing (#12953) --- docs/en/connectors/formats/protobuf.md | 13 ++- docs/en/connectors/source/Kafka.md | 14 +-- docs/zh/connectors/formats/protobuf.md | 6 + docs/zh/connectors/source/Kafka.md | 11 +- .../protobuf/ProtobufDeserializationSchema.java | 8 ++ .../ProtobufDeserializationSchemaTest.java | 129 +++++++++++++++++++++ ...stryAwareProtobufDeserializationSchemaTest.java | 6 +- 7 files changed, 171 insertions(+), 16 deletions(-) diff --git a/docs/en/connectors/formats/protobuf.md b/docs/en/connectors/formats/protobuf.md index 916da551b7..1dd120b301 100644 --- a/docs/en/connectors/formats/protobuf.md +++ b/docs/en/connectors/formats/protobuf.md @@ -160,4 +160,15 @@ sink { plugin_input = "kafka_table" } } -``` \ No newline at end of file +``` + +## Tombstone and empty messages + +A Kafka tombstone is a record whose value is `null`, used on a compacted topic to mark its key as +deleted. The Protobuf source reads such a record as one 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. + +An empty (zero length) value is different. Protobuf defines it as a valid message in which every +field holds its default value, so it is decoded into a row of those defaults rather than being +passed over. diff --git a/docs/en/connectors/source/Kafka.md b/docs/en/connectors/source/Kafka.md index a42a18ce82..183a530470 100644 --- a/docs/en/connectors/source/Kafka.md +++ b/docs/en/connectors/source/Kafka.md @@ -618,13 +618,13 @@ Note：key/value is of type byte[]. 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`, -`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. +`avro`, `compatible_kafka_connect_json` and `protobuf` 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. For +`protobuf` an empty value is a valid message in which every field holds its default value, so it +is decoded into a row of defaults rather than passed over. ### Streaming With Dynamic Partition Discovery and EXACTLY_ONCE Sink diff --git a/docs/zh/connectors/formats/protobuf.md b/docs/zh/connectors/formats/protobuf.md index 5fac7f9321..14c874edbe 100644 --- a/docs/zh/connectors/formats/protobuf.md +++ b/docs/zh/connectors/formats/protobuf.md @@ -162,3 +162,9 @@ sink { } ``` +## 墓碑消息与空消息 + +墓碑消息是只有 key、没有 value 的 Kafka 记录，压缩主题用它标记某个 key 已被删除。Protobuf 源端把这类记录读取为不产生任何数据行的记录： +它不属于格式错误，因此无论 `format_error_handle_way` 取何值都会被跳过，并且位点会继续向前推进。 + +空（长度为 0）的 value 不同。Protobuf 将其定义为每个字段都取默认值的合法消息，因此会被解码为由这些默认值组成的数据行，而不会被跳过。 diff --git a/docs/zh/connectors/source/Kafka.md b/docs/zh/connectors/source/Kafka.md index e238206d2e..dc6b8a4377 100644 --- a/docs/zh/connectors/source/Kafka.md +++ b/docs/zh/connectors/source/Kafka.md @@ -607,12 +607,11 @@ source { ### 墓碑消息（tombstone） 墓碑消息是只有 key、没有 value 的 Kafka 记录，压缩主题用它标记某个 key 已被删除。对于 `json`、`text`、`canal_json`、 -`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` 格式没有这样的空值判断，不适用上述行为。 +`debezium_json`、`ogg_json`、`maxwell_json`、`avro`、`compatible_kafka_connect_json` 和 `protobuf` 格式，源端把墓碑 +消息读取为不产生任何数据行的记录：它不属于格式错误，因此无论 `format_error_handle_way` 取何值都会被跳过，并且位点会继续向前推进。 +其中 `text`、`canal_json`、`debezium_json`、`ogg_json` 和 `avro` 对空（长度为 0）的 value 也做同样处理；`json`、 +`maxwell_json` 和 `compatible_kafka_connect_json` 仅跳过 null value。对于 `protobuf`，空 value 是所有字段都取默认值的 +合法消息，因此会被解码成一行默认值，而不是被跳过。 ### 配合动态分区发现与 EXACTLY_ONCE 下游的流式作业 diff --git a/seatunnel-formats/seatunnel-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchema.java b/seatunnel-formats/seatunnel-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchema.java index 3e3800a59a..ed38b294c2 100644 --- a/seatunnel-formats/seatunnel-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchema.java +++ b/seatunnel-formats/seatunnel-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchema.java @@ -51,6 +51,14 @@ public class ProtobufDeserializationSchema implements DeserializationSchema<SeaT @Override public SeaTunnelRow deserialize(byte[] message) throws IOException { + if (message == 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: protobuf defines it as a valid + // message whose every field holds its default value. + return null; + } Descriptors.Descriptor descriptor = this.converter.getDescriptor(); DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor, message); return convertToRow(dynamicMessage); diff --git a/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchemaTest.java b/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchemaTest.java new file mode 100644 index 0000000000..a86b2a07da --- /dev/null +++ b/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchemaTest.java @@ -0,0 +1,129 @@ +/* + * 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.protobuf; + +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.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.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Covers the format entry point the Kafka source uses when {@code strip_schema_registry_header} is + * left at its default, which is this schema rather than {@link + * SchemaRegistryAwareProtobufDeserializationSchema}. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class ProtobufDeserializationSchemaTest { + + private static final String PROTO_CONTENT = + "syntax = \"proto3\";\n" + + "\n" + + "package org.apache.seatunnel.format.protobuf;\n" + + "\n" + + "option java_outer_classname = \"TombstoneProto\";\n" + + "\n" + + "message TestMessage {\n" + + " int32 id = 1;\n" + + " string name = 2;\n" + + "}"; + + private static final String MESSAGE_NAME = "TestMessage"; + + private ProtobufDeserializationSchema schema; + + @BeforeAll + void setUpSchema() { + schema = new ProtobufDeserializationSchema(createCatalogTable()); + } + + private CatalogTable createCatalogTable() { + Map<String, String> options = new HashMap<>(); + options.put("protobuf_schema", PROTO_CONTENT); + options.put("protobuf_message_name", MESSAGE_NAME); + + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "name"}, + new SeaTunnelDataType<?>[] {BasicType.INT_TYPE, BasicType.STRING_TYPE}); + + CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("tombstone_table", rowType); + catalogTable.getOptions().putAll(options); + return catalogTable; + } + + @Test + void deserializeReturnsNoRowForATombstone() throws IOException { + // A Kafka tombstone delivers a null value. Before the guard this raised a + // NullPointerException from inside DynamicMessage.parseFrom. + Assertions.assertNull(schema.deserialize((byte[]) null)); + } + + @Test + void collectorReceivesNothingForATombstone() throws IOException { + RecordingCollector collector = new RecordingCollector(); + + schema.deserialize((byte[]) null, collector); + + Assertions.assertTrue( + collector.collected.isEmpty(), + "a tombstone must not produce a row, so nothing may reach the collector"); + } + + @Test + void anEmptyPayloadRemainsAValidMessageOfDefaults() throws IOException { + // Protobuf defines a zero length payload as a valid message whose fields all hold + // their defaults, so it must keep decoding rather than being treated as a tombstone. + RecordingCollector collector = new RecordingCollector(); + + SeaTunnelRow row = schema.deserialize(new byte[0]); + schema.deserialize(new byte[0], collector); + + Assertions.assertNotNull(row); + Assertions.assertEquals(0, row.getField(0)); + Assertions.assertEquals("", row.getField(1)); + Assertions.assertEquals(1, collector.collected.size()); + } + + private static class RecordingCollector implements Collector<SeaTunnelRow> { + private final List<SeaTunnelRow> collected = new ArrayList<>(); + + @Override + public void collect(SeaTunnelRow record) { + collected.add(record); + } + + @Override + public Object getCheckpointLock() { + return this; + } + } +} diff --git a/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/SchemaRegistryAwareProtobufDeserializationSchemaTest.java b/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/SchemaRegistryAwareProtobufDeserializationSchemaTest.java index 60f179d54a..6a489d7e3c 100644 --- a/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/SchemaRegistryAwareProtobufDeserializationSchemaTest.java +++ b/seatunnel-formats/seatunnel-format-protobuf/src/test/java/org/apache/seatunnel/format/protobuf/SchemaRegistryAwareProtobufDeserializationSchemaTest.java @@ -149,8 +149,10 @@ public class SchemaRegistryAwareProtobufDeserializationSchemaTest { } @Test - void throwsWhenMessageIsNull() { - Assertions.assertThrows(NullPointerException.class, () -> schema.deserialize(null)); + void returnsNoRowWhenMessageIsNull() throws IOException { + // A null value is a Kafka tombstone, which this schema hands to the inner schema + // unchanged. It yields no row instead of raising a NullPointerException. + Assertions.assertNull(schema.deserialize(null)); } @Test
