sqd opened a new pull request, #333:
URL: https://github.com/apache/flink-connector-kafka/pull/333
## What is the purpose of the change
This pull request makes the Kafka source commit the offsets of the savepoint
when a job is stopped with a savepoint (FLINK-36532).
The source commits the offsets of a completed checkpoint by queueing an
asynchronous commit on the split fetcher. Stop-with-savepoint closes the reader
right after the savepoint completes, and the fetcher shuts down without running
the queued commit, so the savepoint's offsets never reach Kafka. With
`group.protocol=consumer`, closing the consumer also drops a commit that is
already in flight. This happens with and without `--drain`.
The consumer group then lags behind the savepoint. Lag monitoring is wrong,
and a job that starts from the committed offsets instead of the savepoint
re-reads records.
The reader now commits the offsets of the latest completed checkpoint
synchronously when it closes, unless their asynchronous commit has already
finished.
## Brief change log
- `KafkaSourceReader` records the latest completed checkpoint and whether
its asynchronous offset commit has finished. When it closes before that commit
has finished, it hands the checkpoint's offsets to the fetcher manager.
- `KafkaSourceFetcherManager` passes the offsets to the split reader of a
running fetcher. When no fetcher can take them, it starts one for the commit.
- `KafkaPartitionSplitReader` commits the offsets with `commitSync` (10
second timeout) on the fetcher thread before it closes the consumer, and
creates the consumer first if the fetcher has not created it yet. A failed
commit is logged and counted in the failed-commit metric, as on checkpoint, and
does not fail the close.
- Offsets of a checkpoint that did not complete, and of the snapshot that
`DynamicKafkaSourceReader` takes outside a checkpoint, are never committed on
close.
- The "Consumer Offset Committing" section of the Kafka docs, in
`docs/content` and `docs/content.zh`, describes the commit on close.
## Verifying this change
This change added tests and can be verified as follows:
- Added `KafkaSourceStopWithSavepointITCase`, which stops a job with a
savepoint, with and without drain, and checks that every partition's committed
offset matches the savepoint. Periodic checkpoints are off, so the savepoint is
the only checkpoint that can commit. Without the change, no partition has a
committed offset.
- Added tests to `KafkaSourceReaderTest`:
- closing the reader commits the latest completed checkpoint, with both
`group.protocol=classic` and `group.protocol=consumer`;
- closing it does not commit a checkpoint that did not complete;
- closing it does not commit a snapshot taken outside a checkpoint.
- Added tests to `KafkaPartitionSplitReaderTest`:
- the commit on close survives a wakeup that is still pending from
shutting down the fetcher;
- the commit on close creates the consumer when the reader has none;
- offsets handed over after `close()` has started are rejected.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency, including
`kafka.version` or `flink.version`): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)` or `@Experimental`, or are the Table options or the PyFlink
wrappers changed: no
- Checkpointed state, its serializers, or exactly-once delivery (splits,
enumerator state, writer state, committables, transactions): no
- The per-record code paths (split reader, record emitter, writer,
serialization schemas; performance sensitive): no. The split reader changes
only in `close()`.
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
- If the docs changed, are both `docs/content` and `docs/content.zh`
updated? yes
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 5.5)
--
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]
