[FLINK-40660][connector-base] Retry the shutdown synchronization put in SplitFetcher

[MartijnVisser]

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?

• Yes (please specify the tool below)

Generated-by: Claude Code (Claude Opus 5)

[flinkbot]

CI report:

• 44b2193 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build