SEPURI-SAI-KRISHNA commented on issue #12966:
URL: https://github.com/apache/seatunnel/issues/12966#issuecomment-6386059553
Thanks @SEZ9. The PR is already open as #12967, and it does exactly what you
describe: the null value is handled before any format specific parsing runs.
```java
if (msg.value() == 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: it is a decode error
here, as before.
return;
}
```
That sits in `CompatibleKafkaConnectDeserializationSchema.deserialize`,
above `convertToSinkRecord`, so a tombstone never reaches `setRowKind` on a
null row. A zero length value stays a decode error, as you agreed.
**On covering both `fail` and `skip`: I checked, and those modes are not
reachable at this layer.** `format_error_handle_way` is a Kafka source option.
`KafkaSource` passes it to `KafkaRecordEmitter`, and the emitter consults it in
exactly one place, inside `catch (Exception e)` around the deserialize call: on
`SKIP` it logs and continues, otherwise it rethrows. The format module itself
has zero references to `MessageFormatErrorHandleWay`. With the guard in place
`deserialize` returns normally for a tombstone, no exception is thrown, and
neither branch of that catch is entered. So a unit test of the format cannot
select a mode, and after this fix the mode no longer decides the outcome for a
tombstone: both produce no row and the offset advances.
What the two new tests in `CompatibleKafkaConnectDeserializationSchemaTest`
do assert is the behaviour itself: a tombstone produces no row, and a well
formed record is still read with its fields intact.
Worth noting for the `skip` half you mention: on the current code `skip`
cannot work for a tombstone in any format, because the emitter's skip branch
calls `new String(consumerRecord.value())` on the same null value. That is the
emitter side defect tracked in #12926 and fixed by #12927, which is now
approved.
If you would like mode level coverage as well, that belongs in the Kafka
connector IT rather than in the format module, and I am happy to add it once
the emitter fix lands.
--
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]
