This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 2229f1bdc37 [FLINK-40456][datastream] `AsyncWaitOperator` can
permanently drop an element's result when timeout races with a retry
2229f1bdc37 is described below
commit 2229f1bdc379805a7b2e1ad8d5f8aa0fd9e240e3
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Mon Aug 24 17:52:43 2026 +0200
[FLINK-40456][datastream] `AsyncWaitOperator` can permanently drop an
element's result when timeout races with a retry
---
.../api/operators/async/AsyncWaitOperator.java | 18 +++++--
.../api/operators/async/AsyncWaitOperatorTest.java | 60 ++++++++++++++++++++++
2 files changed, 75 insertions(+), 3 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
index 873e7c3ff67..7c2972d8ee8 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
@@ -484,6 +484,9 @@ public class AsyncWaitOperator<IN, OUT>
*/
private final AtomicBoolean retryAwaiting = new AtomicBoolean(false);
+ // set once the timeout fired; makes the result terminal and bypass
the retry path
+ private final AtomicBoolean timedOut = new AtomicBoolean(false);
+
public RetryableResultHandlerDelegator(
StreamRecord<IN> inputRecord,
ResultFuture<OUT> resultFuture,
@@ -511,6 +514,9 @@ public class AsyncWaitOperator<IN, OUT>
// cancel delayed retry timer first
cancelRetryTimer();
+ // timeout result is terminal: route it straight to the
handler, not the retry path
+ timedOut.set(true);
+
// force reset retryAwaiting to prevent the handler to trigger
retry unnecessarily
retryAwaiting.set(false);
@@ -522,7 +528,7 @@ public class AsyncWaitOperator<IN, OUT>
public void complete(Collection<OUT> results) {
Preconditions.checkNotNull(
results, "Results must not be null, use empty collection
to emit nothing");
- if (!retryDisabledOnFinish.get() &&
resultHandler.inputRecord.isRecord()) {
+ if (shouldProcessResultForRetry()) {
processRetryInMailBox(results, null);
} else {
cancelRetryTimer();
@@ -533,7 +539,7 @@ public class AsyncWaitOperator<IN, OUT>
@Override
public void completeExceptionally(Throwable error) {
- if (!retryDisabledOnFinish.get() &&
resultHandler.inputRecord.isRecord()) {
+ if (shouldProcessResultForRetry()) {
processRetryInMailBox(null, error);
} else {
cancelRetryTimer();
@@ -546,7 +552,7 @@ public class AsyncWaitOperator<IN, OUT>
public void complete(CollectionSupplier<OUT> supplier) {
Preconditions.checkNotNull(
supplier, "Runnable must not be null, return empty
collection to emit nothing");
- if (!retryDisabledOnFinish.get() &&
resultHandler.inputRecord.isRecord()) {
+ if (shouldProcessResultForRetry()) {
mailboxExecutor.submit(
() -> {
try {
@@ -563,6 +569,12 @@ public class AsyncWaitOperator<IN, OUT>
}
}
+ private boolean shouldProcessResultForRetry() {
+ return !timedOut.get()
+ && !retryDisabledOnFinish.get()
+ && resultHandler.inputRecord.isRecord();
+ }
+
private void processRetryInMailBox(Collection<OUT> results, Throwable
error) {
mailboxExecutor.execute(
() -> processRetry(results, error), "delayed retry or
complete");
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
index 2c8dc6fb880..e4784392f64 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
@@ -1399,6 +1399,66 @@ public class AsyncWaitOperatorTest {
testProcessingTimeAlwaysTimeoutFunctionWithRetry(AsyncDataStream.OutputMode.UNORDERED);
}
+ /** A timeout firing while a retry is already queued must still emit its
result. */
+ @Test
+ void testTimeoutRaceWithRetry() throws Exception {
+ ControllableExceptionThenTimeoutFunction.releaseCompletion = new
CountDownLatch(1);
+ ControllableExceptionThenTimeoutFunction.completionEnqueued = new
CountDownLatch(1);
+
+ try (OneInputStreamOperatorTestHarness<Integer, Integer> testHarness =
+ createTestHarnessWithRetry(
+ new ControllableExceptionThenTimeoutFunction(),
+ TIMEOUT,
+ 1,
+ AsyncDataStream.OutputMode.UNORDERED,
+ exceptionRetryStrategy)) {
+
+ testHarness.open();
+ testHarness.setProcessingTime(0L);
+
+ // start the async call; its timeout timer is now registered at
TIMEOUT
+ synchronized (testHarness.getCheckpointLock()) {
+ testHarness.processElement(new StreamRecord<>(1, 1L));
+ }
+
+ // let the async call complete exceptionally, then wait until the
resulting retry mail
+ // is actually enqueued in the mailbox
+
ControllableExceptionThenTimeoutFunction.releaseCompletion.countDown();
+
ControllableExceptionThenTimeoutFunction.completionEnqueued.await();
+
+ // fire the timeout while the retry mail is still queued, then run
both mails in order
+ testHarness.setProcessingTime(TIMEOUT + 1L);
+ drainMailbox(testHarness);
+
+ assertThat(testHarness.getOutput()).containsExactly(new
StreamRecord<>(-1, 1L));
+ }
+ }
+
+ private static class ControllableExceptionThenTimeoutFunction
+ extends AlwaysTimeoutWithDefaultValueAsyncFunction {
+ private static final long serialVersionUID = 2L;
+
+ // released by the test to let the async call complete exceptionally
+ static CountDownLatch releaseCompletion;
+ // counted down once the exceptional completion has been enqueued into
the mailbox
+ static CountDownLatch completionEnqueued;
+
+ @Override
+ public void asyncInvoke(Integer input, ResultFuture<Integer>
resultFuture) {
+ tryCounts.merge(input, 1, Integer::sum);
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ releaseCompletion.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ resultFuture.completeExceptionally(new
Exception("Dummy error"));
+ completionEnqueued.countDown();
+ });
+ }
+ }
+
/**
* Test the AsyncWaitOperator with an always-timeout async function under
ordered mode and
* processing time.
