pltbkd commented on PR #28305: URL: https://github.com/apache/flink/pull/28305#issuecomment-5084488324
> Thanks for the PR! The mechanism (chain-level availability folding + deferred emit) looks good. One question on the semantics. > > An async record's life has three steps: > > * (1) enqueue into the async queue (mailbox thread) > * (2) async completion (user thread) > * (3) emit to downstream (mailbox thread) > > IIUC (1) should wait only on queue capacity and (3) only on downstream. The PR gets (3) right via deferred emit, but (1) waits on queue AND downstream (`taskIsAvailable()` ANDs `recordWriter.isAvailable()` with the new provider) — while enqueue never touches downstream, and (3)'s deferral already covers downstream fullness. Side effect: downstream transiently full + empty queue → no pulling, async parallelism drains to zero. > > Would it make sense to let the folded provider _replace_ the recordWriter check instead of being ANDed? The chain wiring is already replacement-style (an operator absorbs its downstream subtree; without one, the folded provider _is_ the record writers); only the StreamTask boundary keeps the AND. Details inline. Thanks for the review! As we've discussed offline, the current PR's semantics seem to have an issue: when the async operator is internally busy (queue is full but downstream has capacity), the waiting path reports 'soft backpressured' instead of 'busy', which is different from the current behavior and may confuse bottleneck survey. For a task, I think backpressure should only be counted when it can't work at full capacity because it can't write out. Based on this, I'm planning the following changes: In `taskIsAvailable()`: when an SBP operator exists, skip `recordWriter.isAvailable()` — the chain provider alone gates input processing, as you suggested. In `processInput()`'s waiting path: when the chain provider is unavailable, still check `recordWriter` to classify the reason: - `recordWriter` also unavailable → backpressure (root cause: downstream). Timer → `softBackPressuredTimePerSecond`. - `recordWriter` available → busy (internal processing, e.g., queue full from pending callbacks). Timer → null → falls into `busyTimeMsPerSecond`. This preserves the ability to distinguish "downstream is the bottleneck" from "operator's internal processing is the bottleneck" for dashboard-based diagnosis. WDYT? Also, since the approach now retains `recordWriter` for classification (not direct replacement), `SupportsOutputAvailability` might not fit well. Would `SupportsChainAvailability` work better, or should we keep the current `SupportsSoftBackpressure` naming? -- 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]
