SEPURI-SAI-KRISHNA opened a new issue, #12952: URL: https://github.com/apache/seatunnel/issues/12952
### Search before asking - [X] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened A Kafka tombstone is a record with a key and a `null` value, used to mark that key as deleted. With `format = protobuf`, such a record fails the job with a `NullPointerException` raised inside the protobuf runtime, because `ProtobufDeserializationSchema.deserialize(byte[])` passes the value straight to `DynamicMessage.parseFrom` without checking it. Every other text-oriented format the Kafka source supports already passes over a tombstone and produces no row. Protobuf is the one remaining format with no guard. This was split out of #12926 / #12927, which fix the same defect for `avro`. The null-safe skip handler added by #12927 means a protobuf tombstone can at least be skipped today when the user opts in, but under the default `format_error_handle_way = fail` it still kills the job, and even under `skip` it is reported as a format error rather than as a record that legitimately produces no row. ### Measured, not assumed Against `protobuf-java` 3.25.3, the version declared by `seatunnel-format-protobuf/pom.xml`, using a descriptor built for `message Person { string name = 1; int32 age = 2; }`: ``` null -> java.lang.NullPointerException: Cannot read the array length because "data" is null at com.google.protobuf.AbstractMessageLite$Builder.mergeFrom(AbstractMessageLite.java:191) at com.google.protobuf.AbstractMessage$Builder.mergeFrom(AbstractMessage.java:505) at com.google.protobuf.DynamicMessage.parseFrom(DynamicMessage.java:95) empty[0] -> returned a message, fields: name="" age=0 serializedSize=0 ``` Two results matter here, and they point in opposite directions: 1. A `null` value throws. This is the tombstone case and it is the defect. 2. A zero length value does **not** throw. Protobuf defines an empty message as a valid message whose fields all hold their defaults, so `new byte[0]` legitimately decodes to `name=""`, `age=0`. This is correct protobuf behaviour and this issue does **not** propose changing it. That is the difference from `avro`, where a zero length buffer is not a valid datum, and it is why the guard proposed below is null only. ### Where the guard is missing `seatunnel-formats/seatunnel-format-protobuf/src/main/java/org/apache/seatunnel/format/protobuf/ProtobufDeserializationSchema.java` on current `dev`: ```java @Override public SeaTunnelRow deserialize(byte[] message) throws IOException { Descriptors.Descriptor descriptor = this.converter.getDescriptor(); DynamicMessage dynamicMessage = DynamicMessage.parseFrom(descriptor, message); return convertToRow(dynamicMessage); } ``` `SchemaRegistryAwareProtobufDeserializationSchema.deserialize` is worth noting because it looks like it handles the case but does not. At its line 69 it recognises the payload: ```java if (message == null || message.length == 0) { return inner.deserialize(message); } ``` and then delegates to the unguarded schema above, so it propagates the same `NullPointerException`. ### The current behaviour is pinned by a test, but was never designed `SchemaRegistryAwareProtobufDeserializationSchemaTest` asserts the crash as the expected result: ```java @Test void throwsWhenMessageIsNull() { Assertions.assertThrows(NullPointerException.class, () -> schema.deserialize(null)); } ``` Worth stating plainly, because it raises the fair question of whether failing on a null value is deliberate. It does not appear to be: - it was added by #12039, `[Fix][Format][Protobuf] Support empty Schema Registry payloads`, where it replaced an older `testDeserializeEmptyMessage`. That PR was about zero length payloads, and this assertion recorded what the code happened to do with a null one. - no option, documentation or error code describes a null value as a protobuf format error. The `NullPointerException` comes from inside the protobuf runtime, not from any SeaTunnel check. - every other format in the table below reached the opposite conclusion. So a fix has to change that assertion, which is worth flagging in advance rather than discovering in review. ### Inconsistent with every other format Verified by reading each file on current `dev`: | format | guard | line | | --- | --- | --- | | `text` | `message == null \|\| message.length == 0` | `TextDeserializationSchema.java:176` | | `canal_json` | `message == null \|\| message.length == 0` | `CanalJsonDeserializationSchema.java:221` | | `debezium_json` | `message == null \|\| message.length == 0` | `DebeziumJsonDeserializationSchema.java:107` | | `ogg_json` | `message == null \|\| message.length == 0` | `OggJsonDeserializationSchema.java:128` | | `json` | `message == null` | `JsonDeserializationSchema.java:129,136` | | `maxwell_json` | `message == null` | `MaxWellJsonDeserializationSchema.java:115` | | `avro` | none on `dev`; null and empty guarded by #12927 | `AvroDeserializationSchema.java` | | `protobuf` | **none** | `ProtobufDeserializationSchema.java:53-57` | ### How it is reached `KafkaRecordEmitter.emitRecord` calls `deserializationSchema.deserialize(consumerRecord.value(), outputCollector)`, and the `DeserializationSchema` default implementation forwards to `deserialize(byte[])`. For a tombstone `consumerRecord.value()` is `null`, so the call above runs with `message == null`. The `NullPointerException` is a `RuntimeException`, so the emitter's `catch (Exception e)` sees it: - `format_error_handle_way = fail`, the default: rethrown, and the job fails on the tombstone. - `format_error_handle_way = skip`: logged at WARN and the offset advances. This path only became usable with #12927; before it, the handler itself dereferenced the null value and raised a second `NullPointerException`, so the offset never advanced and the record was replayed after every restart. ### What you expected to happen A tombstone should produce no row and the offset should advance past it, regardless of `format_error_handle_way`, exactly as it already does for `json`, `text`, `canal_json`, `debezium_json`, `ogg_json` and `maxwell_json`. A deletion marker is not a malformed message, so it should not be routed through format error handling at all. ### Proposed fix Return `null` from `ProtobufDeserializationSchema.deserialize(byte[])` when `message == null`, which the `DeserializationSchema` default contract already treats as "no row to collect". This matches `json` and `maxwell_json`, which guard a null value only. Deliberately **not** in scope: - zero length payloads, which are valid protobuf messages as measured above - any change to `SchemaRegistryAwareProtobufDeserializationSchema` beyond what it inherits from the fixed inner schema - the protobuf descriptor and field resolution work in #12902 and #12660, which touch `CompileDescriptor` and the converters and do not overlap this method ### SeaTunnel Version `dev` ### Note Reported as the follow-up promised on #12927, where @DanielLeens asked that the protobuf lane be linked once opened. ### Are you willing to submit PR? - [X] Yes I am willing to submit a PR! ### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
