SEPURI-SAI-KRISHNA opened a new pull request, #12927: URL: https://github.com/apache/seatunnel/pull/12927
Yes, and thank you for maintaining it: this is a contribution to Apache SeaTunnel. ### Scope agreed before submission Per this repository's contribution workflow, the three questions were put to the developer and answered before any code was written: - **Issue and acceptance criteria**: closes #12926. A Kafka tombstone must produce no row, must not fail the job, and the offset must advance past it, matching the seven other formats that already guard a null value. Rejected alternative: keeping a tombstone fatal for Avro but with a typed exception, which still crash-loops on a legitimate compacted-topic record. - **Modules and compatibility**: `seatunnel-format-avro` and `connector-kafka`. No option, schema or API change. Records that deserialize today are unaffected; the only behaviour change is that a null or empty value now yields no row instead of raising. - **Required validation**: `spotless:apply`, the mandatory `-DskipTests verify`, both module unit runs, a mutation matrix, and an E2E case in the existing `KafkaIT`. ### Duplicate search Searched the open and closed PRs again immediately before opening this one, as the contribution workflow asks. No PR or issue addresses a Kafka tombstone or a null Avro message. Three open PRs touch `seatunnel-format-avro`, and none of them overlaps this defect: - #12900 adds an opt-in nullable Avro ROW layout. It changes the `AvroDeserializationSchema` constructor; this PR changes `deserialize(byte[])`. - #12905 fixes `tablePath.get().toString()` inside `deserializePayload`, about 25 lines below the guard added here. - #12906 fixes `ByteBuffer` boundaries in `AvroToRowConverter`, a file this PR does not touch. Closed #6795 is the same option failing for a different reason: `skip` was not wired up at all for a `NumberFormatException` in the text format. #7838 fixed that by adding the try/catch, and in doing so introduced the `new String(consumerRecord.value())` line that defect 2 is about. The two are complementary rather than duplicates. ### Purpose of this pull request A tombstone, which is a record with a key and a `null` value, fails an Avro Kafka source with a raw `NullPointerException`, and `format_error_handle_way = skip` cannot rescue it because the skip handler throws a second `NullPointerException` on the same null value. The offset is never advanced, so the record is replayed after every restart. Measured by driving `KafkaRecordEmitter.emitRecord` with a real tombstone, split offset starting at 42: | format | handling | before | offset | after this PR | offset | | --- | --- | --- | --- | --- | --- | | json | fail / skip | 0 rows, no error | 43 | unchanged | 43 | | avro | fail | `NPE: ... "message" is null` | 42 | 0 rows, no error | 43 | | avro | skip | `NPE: ... "bytes" is null` | 42 | 0 rows, no error | 43 | The two exception messages name different variables, which is what separates the two defects fixed here. ### Defect 1: `AvroDeserializationSchema` had no tombstone guard `deserializePayload(message, 0, message.length)` dereferences the array before anything checks it. json, text, csv, ogg, canal, debezium and maxwell all guard this, several with the comment `// skip tombstone messages`, which this change reuses verbatim so the formats read alike. This also covers `connector-pulsar`, which builds the same schema. ### Defect 2: the skip handler could not skip a record with no value `new String(consumerRecord.value())` throws when the value is null, so the exception escapes `emitRecord` and `setCurrentOffset` is never reached. The replacement keeps the message body when there is one and otherwise prints `<null>`, and names the topic, partition and offset so a skipped record can actually be located in the log. This half is reachable from any format that raises on a null value, so it is fixed independently of Avro. ### Does this PR introduce any user-facing change? Yes, and it is documented in `docs/en/connectors/source/Kafka.md` and `docs/zh/connectors/source/Kafka.md` under a new "Tombstone records" section. A tombstone, or an empty value, now produces no row instead of failing the job. Nothing that deserializes today changes. This is a bug fix rather than an incompatible change, so there is no entry in `incompatible-changes.md`: the previous behaviour was an uncaught `NullPointerException`, not a documented outcome. ### How was this patch tested? Three new tests, each of which fails without the change: - `AvroSerializationSchemaTest#testDeserializeTombstoneIsSkippedInsteadOfThrowing`, covering a null and an empty value, with and without Confluent header stripping. - `KafkaRecordEmitterTest#emitRecordShouldSkipAnAvroTombstoneAndAdvanceTheOffset`, over both values of `MessageFormatErrorHandleWay`, asserting no row and that the offset advances from 42 to 43. - `KafkaRecordEmitterTest#emitRecordShouldNotFailInsideTheSkipHandlerWhenTheValueIsNull`, which uses a schema that always raises, so it pins the emitter half independently of Avro. Both existing test classes were extended rather than new ones added. E2E, in the existing `KafkaIT`: `testKafkaAvroTombstoneToAssert` produces 50 Avro records, then a tombstone, then 50 more, and asserts the job exits 0 with exactly 100 rows (`MIN_ROW` and `MAX_ROW` both 100). Its config deliberately leaves `format_error_handle_way` at the default `fail`, since a tombstone is not a format error and reading one must not depend on `skip` being configured. **Mutation matrix**, reverting each guard separately to show both are load-bearing: | mutation | avro format test | emitter tests (2) | | --- | --- | --- | | both fixes present | pass | 2 pass | | revert the Avro guard only | **error** | **1 error** | | revert the emitter log only | pass | **1 error** | | revert both (equals `dev`) | **error** | **2 errors** | Commands run locally: ``` ./mvnw spotless:apply ./mvnw -q -DskipTests verify -pl seatunnel-formats/seatunnel-format-avro,seatunnel-connectors-v2/connector-kafka -am ./mvnw test -pl seatunnel-formats/seatunnel-format-avro # 9 tests, 0 failures, 0 errors ./mvnw test -pl seatunnel-connectors-v2/connector-kafka # 121 tests, 0 failures, 0 errors ``` Unit runs are on JDK 17. On JDK 21 four pre-existing `KafkaRecordEmitterTest` cases error out before reaching any assertion, because Mockito cannot mock `ConsumerRecord` on that JDK; this is unrelated to the change and the new tests construct a real `ConsumerRecord` instead of a mock. ### A note on CI The `unit-test` legs are currently red on every PR based on `dev`, including this one. The cause is a pre-existing break tracked by #12874 with the fix in #12875: `JdbcOptionRuleCompletenessTest.testSinkRuleDeclaresAllSinkOptions` fails with `unknown option keys: [numeric_special_value_as_null]`, because #12800 added the completeness test on 2026-10-16 and #12792 added the option to `JdbcCommonOptions` on 2026-10-18 while declaring it only in the source rule. Nothing in this PR reaches `connector-jdbc`. I will rerun once #12875 lands. ### Check list * [x] If any new Jar binary package adding in your PR, please add License Notice according [New License Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md) * [x] If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs * [x] If necessary, please update `incompatible-changes.md` to describe the incompatibility caused by this PR. * [x] If you are contributing the connector code, please check that the following files are updated: 1. Update [plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties) and add new connector information in it 2. Update the pom file of [seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml) 3. Add ci label in [label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml) 4. Add e2e testcase in [seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/) 5. Update connector [plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config) No new connector and no new dependency, so items 1 to 3 and 5 above do not apply; item 4 is the `KafkaIT` case described above. -- 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]
