[jira] [Updated] (FLINK-40604) AsyncWaitOperator bypasses timeout handler when retry result mail runs after the deadline

["David Peacock (Jira)" <ji...@apache.org>]

     [ 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)