SEZ9 commented on issue #12966: URL: https://github.com/apache/seatunnel/issues/12966#issuecomment-6392350775
Thanks for the detailed walkthrough, @SEPURI-SAI-KRISHNA — that answers my question. Placing the `msg.value() == null` guard at the top of `CompatibleKafkaConnectDeserializationSchema.deserialize`, before `convertToSinkRecord`, is the right spot: the tombstone produces no row and never reaches `setRowKind`, and keeping a zero-length payload as a decode error matches what we discussed. Your point on `fail` / `skip` is well taken. Since `format_error_handle_way` is only consulted in the emitter's `catch` block around the deserialize call, and the guard makes `deserialize` return normally for a tombstone, neither branch is entered and the mode no longer influences the outcome. A format-module unit test cannot select the mode, so I agree the two tests you describe (tombstone yields no row; a well-formed record still decodes with intact fields) are the correct coverage at that layer. I also agree that the `new String(consumerRecord.value())` NPE in the emitter's skip branch is a separate, emitter-side defect and belongs in the fix you referenced rather than in this PR. Remaining asks: 1. In the PR, please add a small assertion that a zero-length value still raises the decode error, so the deliberate boundary you describe in the comment is pinned by a test. 2. Once the emitter fix lands, please follow up with the Kafka connector IT you offered: a tombstone under both `skip` and `fail`, asserting no row is emitted and the offset advances so the record is not replayed after a restart. That closes the second half of this issue end to end. With those, I'm happy to move forward on the PR. <!-- streview-comment:2011 --> -- 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]
