SEPURI-SAI-KRISHNA opened a new pull request, #12967:
URL: https://github.com/apache/seatunnel/pull/12967
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 #12966. A Kafka tombstone, a
record with a key and a `null` value, must produce no row on a
`compatible_kafka_connect_json` source and let the offset advance, independent
of `format_error_handle_way`, exactly as the `json` and `maxwell_json` formats
already behave. A zero length value must keep its current behaviour and stay a
decode error, because an empty payload is a different thing from an absent one.
Rejected alternatives: treating empty and null alike, which would silently drop
a malformed payload; and handling it in the connector rather than the format,
which would leave every other caller of this schema exposed.
- **Modules and compatibility**: only
`seatunnel-format-compatible-connect-json`, only
`CompatibleKafkaConnectDeserializationSchema`. No option, schema, wire format,
SPI or checkpoint change. Records that deserialize today are untouched. The
only behaviour change is that a case which previously raised
`NullPointerException` now produces no row, so no `incompatible-changes.md`
entry is required.
- **Required validation**: `./mvnw spotless:apply`, `./mvnw -q -DskipTests
verify`, and the module's tests. Reported below.
### What problem does this PR solve
A tombstone fails the source instead of being passed over as the deletion
marker it is. This is the same defect already fixed for Avro in #12927 and for
Protobuf in #12953, in a third format.
`format_error_handle_way` cannot rescue it, and both values fail differently:
- `fail`, the **default**, stops the job on the first tombstone.
- `skip` raises a second `NullPointerException` inside the skip handler, so
the offset never advances and the record is replayed after every restart.
### Root cause
`deserialize(ConsumerRecord, Collector)` guards `msg == null` but not
`msg.value() == null`. The null value reaches `convertToSinkRecord`, where the
**key** is guarded and 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 becomes a JSON null, and
`convertJsonNode` already returns `null` for it (`:146`). The caller simply
never checks that result, so `row.setRowKind(rowKind)` dereferences null at
`:126` and `:135`.
The class was therefore half prepared for this case already. The fix
completes it.
### The change
A null-value guard at the top of `deserialize`, mirroring the idiom the
sibling formats use:
```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;
}
```
Guarding at the source condition rather than at the two collection sites is
deliberate: it is precise about what a tombstone is, and it does not change how
any other null-valued intermediate is handled.
### Measured evidence
Driven through the real `KafkaRecordEmitter`, with the schema constructed
exactly as `KafkaSourceConfig.java:392` constructs it, both schema-enable flags
at their default of `true`:
| Case | Mode | Before | After |
|---|---|---|---|
| Well formed record | `fail` | 1 row, offset advances | unchanged |
| Tombstone | `fail` | `NullPointerException` on `setRowKind`, offset stuck
| no row, offset advances |
| Tombstone | `skip` | `NullPointerException` on `new String(value)`, offset
stuck | no row, offset advances |
The well formed control passes in the same run, so the failures belong to
the tombstone and not to the fixture.
### Tests
The module had no test sources at all, so
`CompatibleKafkaConnectDeserializationSchemaTest` is a new class with the ASF
header, rather than a new class added beside an existing one.
- `deserializeProducesNoRowForATombstone`
- `deserializeStillReadsAWellFormedRecord`
**Mutation checked.** With the guard removed, the tombstone test fails with
the production `NullPointerException` and the well formed test stays green, so
the coverage is tied to this defect and not decorative.
### Validation
- `./mvnw spotless:apply` run, and `spotless:check` passes as part of the
build rather than only `apply`
- `./mvnw -q -DskipTests verify -pl
seatunnel-formats/seatunnel-format-compatible-connect-json`: passes
- `./mvnw test -pl
seatunnel-formats/seatunnel-format-compatible-connect-json`: **2 tests, 0
failures, 0 errors**, `BUILD SUCCESS`
- Markdown is excluded from Spotless in the root `pom.xml`, so the two doc
files need no format pass
### Scope notes
- **Empty payload left alone.** A zero length value is not a tombstone and
remains a decode error.
- **No E2E added.** This format has no existing IT of its own, and a
tombstone is already exercised end to end for Avro by
`KafkaIT#testKafkaAvroTombstoneToAssert` in #12927. Happy to add one if you
would rather see it here.
- **Docs.** `docs/en` and `docs/zh` for this format both gain a note under
`Limitations` describing the tombstone behaviour and distinguishing it from a
zero length value. The Kafka source doc also carries a sentence naming the
formats that pass a tombstone over, but that paragraph is introduced by #12927
and does not exist on `dev` yet, so it cannot be edited from this branch. To
avoid three open PRs editing one paragraph, whichever of #12927, #12953 and
this PR merges last will carry that wording update.
- **Overlap.** #12905 edits two lines inside this same `deserialize` method,
one in each branch (`tablePath.toString()` to `tablePath.get().toString()`),
and #12602 edits both of the doc files touched here. I tested both merges with
`git merge-tree` and neither conflicts today, but whichever lands second may
need a trivial rebase.
--
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]
