SAlexandru opened a new pull request, #29218:
URL: https://github.com/apache/flink/pull/29218
## 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](https://issues.apache.org/jira/browse/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?
- [X] Yes
Generated-by: OpenAI Codex (GPT-5)
(also used Claude Code Opus 5 for review, to try it out)
--
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]
