[
https://issues.apache.org/jira/browse/FLINK-40604?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
David Peacock updated FLINK-40604:
----------------------------------
Description:
`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()`.
I have a fix ready and can submit the PR if the approach has consensus, could a
committer please assign FLINK-40604 to me?
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.
was:
`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.
> 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
> Priority: Major
>
> `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()`.
> I have a fix ready and can submit the PR if the approach has consensus, could
> a committer please assign FLINK-40604 to me?
> 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)
