SEPURI-SAI-KRISHNA opened a new issue, #12966: URL: https://github.com/apache/seatunnel/issues/12966
### 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, a record with a key and a `null` value, fails a Kafka source configured with `format = compatible_kafka_connect_json` instead of being passed over as a deletion marker. This is the same defect already reported for the Avro format in #12926 and for Protobuf in #12952, in a third format. `format_error_handle_way` cannot rescue it, and both of its values fail in a different way: - `fail`, which is the **default**, stops the job on the first tombstone. - `skip` raises a second `NullPointerException` inside the skip handler itself, so the offset is never advanced and the record is replayed after every restart. ### SeaTunnel Version dev (`6b63aca17`) ### SeaTunnel Config ```conf source { Kafka { topic = "test_topic" bootstrap.servers = "localhost:9092" format = compatible_kafka_connect_json # key_converter_schema_enabled and value_converter_schema_enabled both default to true # format_error_handle_way defaults to fail } } ``` ### Running Command ```shell ./bin/seatunnel.sh --config ./config/kafka_connect_json.conf -e local ``` ### Error Exception With the default `format_error_handle_way = fail`: ```log java.lang.NullPointerException: Cannot invoke "org.apache.seatunnel.api.table.type.SeaTunnelRow.setRowKind(org.apache.seatunnel.api.table.type.RowKind)" because "row" is null at CompatibleKafkaConnectDeserializationSchema.deserialize(CompatibleKafkaConnectDeserializationSchema.java:135) ``` With `format_error_handle_way = skip`: ```log java.lang.NullPointerException: Cannot read the array length because "bytes" is null at KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:100) ``` ### Measured behaviour Driven through the real `KafkaRecordEmitter`, with the schema built exactly as `KafkaSourceConfig` builds it (`KafkaSourceConfig.java:392`, both schema-enable flags at their default of `true`): | Case | Mode | Result | Offset advanced | |---|---|---|---| | Well formed record | `fail` | 1 row, `SeaTunnelRow{kind=+I, fields=[1, a]}` | yes, to 11 | | Tombstone | `fail` | `NullPointerException` on `setRowKind` | **no** | | Tombstone | `skip` | `NullPointerException` on `new String(value)` | **no** | The well formed control passes in the same run, so the failures belong to the tombstone and not to the fixture. ### Root cause Two independent points on the same path. **1. The format has no tombstone guard.** `deserialize(ConsumerRecord, Collector)` guards `msg == null` but not `msg.value() == null` (`CompatibleKafkaConnectDeserializationSchema.java:110`). The null value reaches `convertToSinkRecord`, where the key is guarded but the value is not: ```java SchemaAndValue keyAndSchema = (msg.key() == null) ? SchemaAndValue.NULL : keyConverter.toConnectData(msg.topic(), msg.headers(), msg.key()); SchemaAndValue valueAndSchema = valueConverter.toConnectData(msg.topic(), msg.headers(), msg.value()); ``` `JsonConverter.toConnectData` answers a null value with `SchemaAndValue.NULL`, the payload node is then a JSON null, and `convertJsonNode` correctly returns `null` for it (`:146`). The caller never checks that result, so `row.setRowKind(rowKind)` dereferences null at `:126` and `:135`. **2. The skip handler cannot skip it.** `KafkaRecordEmitter` catches the failure and, under `SKIP`, logs `new String(consumerRecord.value())` on the same null value. That is the emitter half already reported in #12926 and fixed in #12927; once that merges, `skip` will work for this format too. The `fail` default will still break, because the format itself has no guard. ### What you expected to happen A tombstone produces no row and the source moves past it, as the `json` and `maxwell_json` formats already do, independent of `format_error_handle_way`. A zero length value should keep its current behaviour and remain a decode error, since it is a different thing from a null value. ### Are you willing to submit a 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]
