SEPURI-SAI-KRISHNA opened a new pull request, #13010:
URL: https://github.com/apache/seatunnel/pull/13010
Yes, contributing to Apache SeaTunnel.
Adds the Kafka connector IT that #12966 asked for. The format fix itself
merged as #12967 (`76e11cb80`), which closed that issue, so this PR carries no
closing keyword: it is the remaining E2E half, kept separate because it is an
E2E change rather than a format module change.
## What this adds
One test method on the existing `KafkaFormatIT`, which is the class that
already covers `compatible_kafka_connect_json` end to end through
`testFormatCompatibleCheck`, plus one job config.
`testFormatCompatibleTombstoneCheck`:
1. Publishes the envelopes the existing `/compatible/compatible_data.txt`
fixture already carries, with a tombstone (a record with a key and a `null`
value) inserted before the last one. The fixture loop is `while ((line =
br.readLine()) != null)`, publishing each line as the record value, so a
tombstone cannot be expressed in the file at all and needs an explicit `new
ProducerRecord<>(topic, "tombstone-key", null)`.
2. Runs the job twice, once with `format_error_handle_way = FAIL` (the
default) and once with `SKIP`.
3. Asserts the tombstone produced no row and every real record still
arrived, by reusing `checkCompatibleFormat()` unchanged. Because the tombstone
is inserted before the last envelope, records on both sides of it have to
arrive for that expectation to hold.
4. Asserts the consumer group committed an offset past the tombstone. This
is the point of the test: a row count alone cannot tell "read correctly" from
"read correctly and then replayed on the next restart", and the replay is the
failure #12966 is about.
## Reviewer notes
**It exercises the same guard path as the unit tests.** The config sets
`format = COMPATIBLE_KAFKA_CONNECT_JSON`. `KafkaSourceConfig` maps that case to
`new CompatibleKafkaConnectDeserializationSchema(...)`, which is the same class
`CompatibleKafkaConnectDeserializationSchemaTest` constructs directly. The
guard is the first thing `deserialize` does with the record value, before
`convertToSinkRecord`, so both the IT and the unit tests enter that one branch.
I stated that as a falsifiable prediction before the guard merged, and then
tested it rather than leaving it as an argument. See the mutation evidence
below.
**The test was mutation checked against the real `dev`, by CI, not by
argument.** Before #12967 merged I predicted that this IT must fail without the
guard, and specifically on the `FAIL` arm only, because #12927 had already made
the emitter's `SKIP` branch survive a null value. The first push of this branch
was based on a `dev` that did not yet carry the guard, so that push was the
guard-absent arm. It came back red, on both JDK 11 and 17, exactly as predicted:
```
testFormatCompatibleTombstoneCheck FAILED at KafkaFormatIT.java:508
org.opentest4j.AssertionFailedError: format_error_handle_way=FAIL: ...
testFormatCompatibleCheck successfully run
```
Line 508 is the job exit code assertion, and the assertion message is
prefixed with the arm, so the failure is located precisely: the `FAIL` arm
aborted the job because the tombstone was treated as a format error. The
pre-existing `testFormatCompatibleCheck` passed in the same run, which rules
out a broken harness or a bad fixture.
That is the half of the pair that is easy to get wrong and easy to skip. A
test that passes tells you very little on its own; this one is now known to be
sensitive to the exact line it is meant to cover.
**Why it is restricted to Zeta.** The committed offset assertion needs
offsets to be committed when a batch job finishes. `KafkaSourceReader` commits
from `notifyCheckpointComplete`, and the repository already records that the
other engines do not reach it in batch:
`KafkaIT.testSourceKafkaStartConfigWithCommitOffset` carries `disabledReason =
"flink and spark won't commit offset when batch job finished"`. The same
annotation and reason are reused here. The guard is in the format module and is
engine independent, so the Zeta run covers it.
**Why its own topic and consumer group.** The broker is shared by every
template run in the class. Putting a tombstone on `jdbc_source_record` would
change the fixture `testFormatCompatibleCheck` reads, and a shared consumer
group would carry committed offsets in from the other lanes and make the offset
assertion meaningless.
**No doc change.** This is test only. The user facing behaviour is
documented by #12967, which carries the Kafka source page update in both
`docs/en` and `docs/zh`.
## Tests
- `./mvnw spotless:check -pl
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e` passes.
- `./mvnw install -pl
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e -am -DskipTests`
is a full module BUILD SUCCESS, with `testCompile` emitting a fresh
`KafkaFormatIT.class`. That proves every Kafka admin and producer API the new
helpers use actually exists, which is the part a reviewer cannot check by
reading.
- The new job config parses: it loads through the same HOCON parser the
runtime uses and is structurally identical to the existing compatible config,
with `${topic}`, `${consumerGroup}` and `${errorHandleWay}` carried through as
substitutable values.
- The IT itself cannot be run locally here, because this account has no
Docker socket permission. I am flagging that rather than glossing over it. CI
is therefore the real signal, and the guard-absent arm above is the reason I am
reasonably confident in it rather than merely hopeful: the test has already
been observed failing for the right reason, on the right arm, at the right line.
- This PR's own CI is the guard-present arm. Its base now carries #12967, so
`testFormatCompatibleTombstoneCheck` is expected to pass here. I will report
that result on this PR rather than assume it.
--
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]
