MartijnVisser opened a new pull request, #318:
URL: https://github.com/apache/flink-connector-kafka/pull/318
## What is the purpose of the change
A bounded `KafkaSource` never finishes once the job has been restored from a
savepoint or a retained
checkpoint, or after a JobManager failover. All three recreate the
enumerator from state.
The recreated enumerator runs its one-time discovery, finds every subscribed
partition already in
`assignedSplits`, and `checkPartitionChanges` returns on the empty
`PartitionChange` before it
reaches `handlePartitionSplitChanges`, which is the only place that sets
`noMoreNewPartitionSplits`.
The restored readers consume up to their stopping offset and then wait
forever for a
`NoMoreSplitsEvent`, which the runtime never re-sends on its own. The same
early return hangs a
bounded source that subscribes to no existing partition on a fresh start.
The early return came in with `0b38d1a9` (FLINK-22147). Before that
refactor, discovery and split
initialisation shared one callable whose callback always ran, so a restored
bounded source did
signal correctly. The Pulsar enumerator does not have this bug, because its
split assigner marks the
discovery finished on every discovery rather than only on a change.
## Brief change log
- `KafkaSourceEnumerator.checkPartitionChanges` lets an empty partition
change through when a
bounded source has run its only discovery, so that it reaches
`handlePartitionSplitChanges` and
signals the registered readers.
- Two test-only `[hotfix]` commits first, each independently green: the
`createEnumerator` helper in
`KafkaSourceEnumeratorTest` dropped its `initialDiscoveryFinished`
argument, and the context proxy
in `DynamicKafkaSourceEnumeratorTest` passed `null` where production passes
`DynamicKafkaSourceEnumerator::handleNoMoreSplits`, so no test could
observe that wiring.
- New `KafkaSourceBoundedRestoreITCase`, plus enumerator coverage for the
static and the dynamic
source.
Periodic discovery and unbounded sources keep returning early. Neither can
act on an empty change,
and treating one as a finished discovery would make the partitions that a
later discovery finds
resolve against the earliest offset rather than the configured one.
Signalling from `addReader` on the strength of the restored
`initialDiscoveryFinished` flag would
need no new state, but it finishes a reader before the post-restore
discovery assigns it a partition
created while the job was down. That is what apache/flink#21909 was rejected
for, and
`testRestoredBoundedEnumeratorAssignsBeforeItSignals` pins the ordering that
avoids it. The
reasoning is written up on the Jira, answering the question left open on
FLINK-33466 in 2023.
## Verifying this change
Flink 2.2.1, JDK 17. Each of the following fails without the production
change and passes with it:
-
`KafkaSourceEnumeratorTest.testRestoredBoundedEnumeratorSignalsNoMoreSplitsWithoutPartitionChanges`
-
`KafkaSourceEnumeratorTest.testBoundedSourceWithoutPartitionsSignalsNoMoreSplits`
-
`DynamicKafkaSourceEnumeratorTest.testBoundedRestoreSignalsNoMoreSplitsWithoutPartitionChanges`
-
`KafkaSourceBoundedRestoreITCase.testBoundedSourceFinishesAfterRestoreFromSavepoint`,
which times
out after two minutes on the base and passes in about six seconds with the
fix
Two further unit tests pass on both sides by design, because they pin
decisions rather than reproduce
the bug: `testUnboundedSourceKeepsSkippingAnEmptyPartitionChange` fails if
the condition is ever
widened past bounded sources, and
`testRestoredBoundedEnumeratorAssignsBeforeItSignals` fails under
the `addReader` variant described above.
`mvn verify` on the module with no skips: 443 unit tests and 326 integration
tests, no failures,
spotless, checkstyle, enforcer and japicmp clean, plus `apache-rat:check` at
the root. The new
ITCase was run 20 consecutive times, all passing. No existing test restored
a bounded `KafkaSource`,
which is the coverage asked for on the earlier attempt in #71 and never
added.
## Does this pull request potentially affect one of the following parts
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: **no**
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery: **yes**. It changes when
`NoMoreSplitsEvent` is sent
to the readers of a bounded source after the enumerator is recreated from
state. No state format
change; `noMoreNewPartitionSplits` stays derived rather than checkpointed.
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no**
- If yes, how is the feature documented? **not applicable**. The documented
behaviour, that a bounded
source exits once all partitions reach their stopping offsets, is what
this restores.
## AI disclosure
- [x] This pull request was written with the help of an AI tool.
Generated-by: Claude Code 2.1.267 (Claude Opus 5). The diagnosis, the fix
and the tests were
reviewed and verified locally by the submitter.
--
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]
