David Peacock created FLINK-40604:
-------------------------------------
Summary: AsyncWaitOperator bypasses timeout handler when retry
result mail runs after the deadline
Key: FLINK-40604
URL: https://issues.apache.org/jira/browse/FLINK-40604
Project: Flink
Issue Type: Bug
Components: API / DataStream
Reporter: David Peacock
`AsyncWaitOperator` can forward a stale retry-attempt exception and fail the
task after the overall async timeout has already elapsed, instead of invoking
the configured `AsyncFunction#timeout` handler.
The race is:
1. An async attempt completes exceptionally before the overall deadline.
2. `RetryableResultHandlerDelegator#completeExceptionally` determines that the
result should be evaluated for retry and queues `processRetry` in the mailbox.
3. Processing time advances past the overall timeout before that retry mail is
processed, while the timeout callback has not run yet.
4. `processRetry` evaluates `!isTimeout()` as false and falls into its generic
terminal branch.
5. That branch forwards the stale attempt error to
`ResultHandler#completeExceptionally`, which calls `failExternally()`.
As a result, the outcome depends on mailbox/timer ordering. A user-provided
timeout handler is bypassed even though the overall operation has timed out.
The async I/O documentation defines the timeout as the maximum duration from
the first invocation through final completion, including retries. It also says
that users can override `AsyncFunction#timeout` to handle an expired request.
The expired retry-result path should therefore use the same timeout handling
path as the timer callback.
h3. Expected behavior
When retry-result mail is processed after the overall async deadline:
- no further retry is scheduled;
- `AsyncFunction#timeout` is invoked exactly once through the existing timeout
path; and
- the timeout handler decides whether to emit a fallback, emit nothing, or
complete exceptionally.
h3. Actual behavior
The stale attempt result is completed normally or exceptionally. For an
exceptional attempt, this can fail and restart the task without invoking
`AsyncFunction#timeout`.
h3. Reproducer
A deterministic `AsyncWaitOperatorTest` can reproduce the race without sleeps:
1. Start an async operation with retries and a custom timeout handler that
emits `-1`.
2. Complete the first attempt exceptionally so retry evaluation is queued in
the mailbox.
3. Quiesce the processing-time service so the timeout callback does not run
first.
4. Advance processing time to `timeout + 1`.
5. Process the queued retry mail.
On unmodified master at `4e9d5412d3dcd4776b750e8a94ba6e86d2b409c6`, the test
errors with:
{code:java}
java.lang.UnsupportedOperationException: MockEnvironment does not support
external task failure. at MockEnvironment.failExternally(...) at
AsyncWaitOperator$ResultHandler.completeExceptionally(...) at
AsyncWaitOperator$RetryableResultHandlerDelegator.processRetry(...) {code}
With the proposed fix, the timeout handler is called once, `-1` is emitted, and
the task is not failed externally.
h3. Proposed fix
At the beginning of `processRetry`, after its existing duplicate-call guard,
check whether the overall timeout has elapsed. If it has, invoke the existing
`timerTriggered()` method and return before evaluating retry predicates or
forwarding the attempt result.
Because `timerTriggered()` may propagate an exception from a user timeout
handler, narrow the `CollectionSupplier` overload's existing `try/catch` to
cover only `supplier.get()`. This prevents a timeout-handler exception from
being reclassified as a supplier failure and then ignored by the duplicate-call
guard.
This is deliberately narrow:
- retry behavior before the deadline is unchanged;
- Flink's existing strict `>` timeout boundary is unchanged;
- no public API or serializer changes are involved; and
- the timeout state transition remains centralized in `timerTriggered()`.
h3. Related issues
- FLINK-30477 prevents retries from continuing after a timeout.
- FLINK-40456 fixes the opposite ordering, where the timeout callback runs
while retry mail is already queued and the timeout result could be dropped.
- FLINK-34501 concerns behavior after retry exhaustion; this issue concerns
retry-result mail processed after the overall timeout.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
