[FLINK-40657][connector-base] Release per-fetcher state in FutureCompletingBlockingQueue

[SAlexandru]

What is the purpose of the change

FutureCompletingBlockingQueue retained a condition and wakeup flag for each
producer ID that created queue state. SplitFetcherManager assigns monotonically
increasing IDs that are never reused, so this state could grow indefinitely as
fetchers were created and shut down.

This change releases per-producer state when its fetcher terminates, bounding
retained state by active or in-flight producers. It preserves sticky wakeups,
prevents release from stranding a producer inside put(), and leaves the
non-full put() path unchanged.

Brief change log

• Replace the producer-indexed ConditionAndFlag array with a lock-guarded map.
• Add releaseProducer() to remove state after a producer permanently finishes.
• Invoke releaseProducer() from the existing SplitFetcherManager shutdown
  hook.
• Keep wakeUpPuttingThread() creating state so wakeup requests remain sticky.
• Track logical waiting putters so state cannot be released while a producer is
  parked or has been signalled but has not yet reacquired the queue lock.
• Clear SplitFetcher.runningTask when task execution fails.
• Add lifecycle, bounded-storage, sparse-ID, wakeup, and concurrency regression
  tests.

Why SplitFetcher.runningTask must be cleared on failure

releaseProducer() requires the shutdown hook to be the producer's final
interaction with the queue.

Previously, SplitFetcher.runOnce() cleared runningTask only when
task.run() returned normally. If a FetchTask failed while retaining a batch,
the following shutdown ordering was possible:

This requires the failure to occur after FetchTask has obtained or retained a
batch. A splitReader.fetch() failure that occurs before returning a batch leaves
lastRecords null and cannot recreate queue state through this path.

runOnce() now clears runningTask in a finally block while holding the
fetcher lock. Successful result processing remains atomic with that clear. A
separate completion flag distinguishes a normal false return from a thrown
exception.

This makes both possible shutdown orderings safe:

• If shutdown() acquires the fetcher lock before runOnce() performs its
  cleanup, it may wake the still-published task. runOnce() then clears the task,
  and the shutdown hook releases any queue state afterward.
• If runOnce() acquires the fetcher lock first, it clears the failed task.
  shutdown() subsequently observes no running task and performs no late
  wakeup.

In both cases, releaseProducer() remains the final interaction with the queue
for that producer ID.

This exceptional path is not the normal split-churn leak reported in
FLINK-40657. It was identified
while ensuring that the new terminal-release invariant also holds during failed
task shutdown.

Failure propagation remains unchanged: Exception instances are wrapped in a
RuntimeException, and Error instances propagate unchanged. The successful
path performs the same single lock reacquisition as before.

Verifying this change

Tests cover:

• Repeated creation and shutdown of real fetchers with monotonically increasing
  IDs.
• Bounded per-producer state across fetcher lifecycles.
• Sparse producer IDs without storage proportional to the largest ID.
• Idempotent release and producer-state isolation.
• Sticky wakeups issued before a producer blocks.
• Release attempts while a producer is parked.
• Cleanup after a waiting producer is interrupted.
• The race where a signalled producer has not yet reacquired the queue lock.
• Clearing runningTask after both Exception and Error failures.

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; implementation changes only, with no API signature
  changes
• The serializers: no
• The runtime per-record code paths (performance sensitive): no; producer state
  is accessed only when a queue is full or explicitly woken, and the non-full
  put() path remains unchanged
• Anything that affects deployment or recovery: 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

Generated-by: OpenAI Codex (GPT-5)
(also used Claude Code Opus 5 for review, to try it out)

[flinkbot]

CI report:

• 42439c0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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