MartijnVisser opened a new pull request, #29182:
URL: https://github.com/apache/flink/pull/29182
## What is the purpose of the change
`SplitFetcher#run` hands over an empty synchronization batch after its loop
ends, so the `SplitReader` is only closed once the records it emitted have been
processed. It discards the return value of that `put`.
`FutureCompletingBlockingQueue#put` returns false without enqueueing when the
queue is full and the fetcher has an unconsumed wakeUp flag, so the batch is
dropped and `recycle()` never runs.
The fetcher then blocks on `recordsProcessedLatch` forever, its
`SplitReader` is never closed and `fetchersToShutDown` is never decremented.
`maybeShutdownFinishedFetchers` has already removed it from the fetcher map, so
`SplitFetcherManager#close` cannot release it either and blocks until its
timeout. The retry terminates because a refused put consumes the flag, and
`runningTask` is already null at that point so nothing can set it again.
## Brief change log
* Retry the shutdown synchronization put in `SplitFetcher#run` instead of
discarding its result
## Verifying this change
This change added tests and can be verified as follows:
* `SplitFetcherTest#testShutdownSynchronizationBatchIsRetriedAfterWakeUp`
fills a capacity-1 queue from another producer, leaves an unconsumed wakeUp for
the fetcher's own index and shuts the fetcher down. It times out on the parent
commit with "The fetcher should have enqueued its shutdown synchronization
batch" and passes with this change.
`release-1.20` is not affected, since it has no `recordsProcessedLatch` and
no synchronization batch. The patched block is byte-identical on `release-2.0`
through `release-2.3`.
## 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)`: yes, `SplitFetcher` is `@PublicEvolving`, but no signature
changes and the behaviour change is confined to its shutdown path
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- 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
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 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]
