ricky2129 opened a new pull request, #12893:
URL: https://github.com/apache/seatunnel/pull/12893
### Purpose of this pull request
With the default `semantics = NON`, the Kafka sink drops records silently
when a send fails.
`KafkaNoTransactionSender.send()` calls `producer.send(record)` without a
callback, and `snapshotState()` only calls `producer.flush()`, which waits for
pending sends but does not throw for failed ones. A record that fails
asynchronously is therefore lost, and the checkpoint still completes past it,
so it is never re-sent. The job keeps running and nothing above the Kafka
client's own WARN lines is logged.
Common causes:
- the topic does not exist and `auto.create.topics.enable=false`
(`TimeoutException: Topic ... not present in metadata after 60000 ms`)
- a record larger than `max.request.size` (`RecordTooLargeException`; in
this case not even a WARN is logged)
- authorization failures
- `delivery.timeout.ms` expiring
#11541 fixed the same pattern for `EXACTLY_ONCE` in
`KafkaTransactionSender`. This PR applies it to the default sender:
- `send()` registers a callback that keeps the first asynchronous failure;
later failures are logged as suppressed.
- `send()` and `snapshotState()` rethrow it as `KAFKA-09`
(`PRODUCE_DATA_FAILED`, added in #11541), so the task fails before the
checkpoint can move past the lost record, and the job restores from the last
completed checkpoint.
Retriable errors such as a leader change are still retried inside the
producer; the callback only sees the final failure.
### Does this PR introduce _any_ user-facing change?
Yes. Under `semantics = NON`, a failed send now fails the task with
`KAFKA-09` instead of silently dropping the record. After the cause is fixed
(for example the topic is created), the job resumes from the last completed
checkpoint and re-sends, so no record is lost; records after that checkpoint
can be delivered twice, which matches the at-least-once contract. A permanent
cause (a record that is always too large, a missing ACL) now keeps the job
failing until it is fixed, instead of skipping the record. The Kafka sink docs
and `incompatible-changes.md` (en/zh) describe this.
### How was this patch tested?
- New `KafkaNoTransactionSenderTest` (`MockProducer`): successful sends
checkpoint; a failed send fails `snapshotState()` with the original cause and
`KAFKA-09`; a failed send rejects later records; the first failure is kept as
the cause. Removing either check, the callback, or the first-failure guard
makes a test fail.
- `connector-kafka` unit tests: 123/123 pass.
- Manual end-to-end check on a 2.3.12-based build with MySQL-CDC -> Kafka
(`compatible_debezium_json` and `json`), Kafka 3.7 with
`auto.create.topics.enable=false`, comparing the unpatched and patched
connector on the same binlog:
- Normal load (27,500 changes, two tables): identical message counts,
keys, partitions and payloads; same throughput.
- Broker restart during streaming: no task failure and no loss on either
build.
- Missing topic: unpatched drops the record and keeps running; patched
fails with `KAFKA-09`, and after the topic is created and the job is restored
(`-r <jobId>`) the record is delivered.
- `max.request.size` smaller than one record: unpatched drops it with no
log line and delivers the next record; patched fails with
`RecordTooLargeException` and does not skip it.
- Savepoint from the unpatched build restored on the patched build and
back again: no records lost.
### Check list
* [ ] 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.
* [ ] 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)
--
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]
