[
https://issues.apache.org/jira/browse/FLINK-40456?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sergey Nuyanzin updated FLINK-40456:
------------------------------------
Description:
When async retries are enabled and an element also has a timeout configured,
the operator can lose that element's result. The element is then never emitted
and never fails — it stays "in flight" forever, so the job (or a test) hangs
indefinitely.
When an element times out, the operator is supposed to emit the timeout's
default result and stop. But that timeout result is passed through the retry
code path, which begins with a guard meaning "a retry is already in progress
for this
element — ignore this completion." If a retry for the same element was just
scheduled or is being processed at that moment, the timeout result hits that
guard and is silently discarded. By then the timeout handler has already
cancelled the
element's retry timer, and the element was never marked as completed — so
nothing remains that could ever complete it. The element is stuck permanently.
Steps to reproduce
{noformat}
docker run --rm --cpus=2 -v {Path-to-flink-local-repo}:/flink -v
{path-to-m2}:/root/.m2 -w /flink -e JAVA_HOME=/usr/lib/jvm/jdk-25.0.3+9
apache/flink-ci-docker:java_8_11_17_21_25_maven_386_jammy ./mvnw -o -q -pl
flink-streaming-java -Phadoop3-tests,hive3 -Djdk25 -Pjava25-target
-Dsurefire.failIfNoSpecifiedTests=false
-Dtest='AsyncWaitOperatorTest#testProcessingTimeWithTimeoutFunctionUnorderedWithRetry'
surefire:test
{noformat}
it fails as
{noformat}
[ERROR]
org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorTest.testProcessingTimeWithTimeoutFunctionUnorderedWithRetry
-- Time elapsed: 100.1 s <<< ERROR!
java.util.concurrent.TimeoutException:
testProcessingTimeWithTimeoutFunctionUnorderedWithRetry() timed out after 100
seconds
at
java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:511)
at
java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1450)
at
java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:2019)
at
java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:187)
Suppressed: java.lang.InterruptedException: sleep interrupted
at java.base/java.lang.Thread.sleepNanos0(Native Method)
at java.base/java.lang.Thread.sleepNanos(Thread.java:509)
at java.base/java.lang.Thread.sleep(Thread.java:540)
at
org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorTest.testProcessingTimeAlwaysTimeoutFunctionWithRetry(AsyncWaitOperatorTest.java:1444)
at
org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorTest.testProcessingTimeWithTimeoutFunctionUnorderedWithRetry(AsyncWaitOperatorTest.java:1399)
at java.base/java.lang.reflect.Method.invoke(Method.java:565)
... 4 more
{noformat}
was:
When async retries are enabled and an element also has a timeout configured,
the operator can lose that element's result. The element is then never emitted
and never fails — it stays "in flight" forever, so the job (or a test) hangs
indefinitely.
When an element times out, the operator is supposed to emit the timeout's
default result and stop. But that timeout result is passed through the retry
code path, which begins with a guard meaning "a retry is already in progress
for this
element — ignore this completion." If a retry for the same element was just
scheduled or is being processed at that moment, the timeout result hits that
guard and is silently discarded. By then the timeout handler has already
cancelled the
element's retry timer, and the element was never marked as completed — so
nothing remains that could ever complete it. The element is stuck permanently.
Steps to reproduce
{noformat}
docker run --rm --cpus=2 -v {Path-to-flink-local-repo}:/flink -v
{path-to-m2}:/root/.m2 -w /flink -e JAVA_HOME=/usr/lib/jvm/jdk-25.0.3+9
apache/flink-ci-docker:java_8_11_17_21_25_maven_386_jammy ./mvnw -o -q -pl
flink-streaming-java -Phadoop3-tests,hive3 -Djdk25 -Pjava25-target
-Dsurefire.failIfNoSpecifiedTests=false
-Dtest='AsyncWaitOperatorTest#testProcessingTimeWithTimeoutFunctionUnorderedWithRetry'
surefire:test
{noformat}
> AsyncWaitOperator can permanently drop an element's result when a timeout
> races with a retry
> --------------------------------------------------------------------------------------------
>
> Key: FLINK-40456
> URL: https://issues.apache.org/jira/browse/FLINK-40456
> Project: Flink
> Issue Type: Sub-task
> Components: API / DataStream, Build System / CI
> Reporter: Sergey Nuyanzin
> Assignee: Sergey Nuyanzin
> Priority: Major
>
> When async retries are enabled and an element also has a timeout configured,
> the operator can lose that element's result. The element is then never
> emitted and never fails — it stays "in flight" forever, so the job (or a
> test) hangs
> indefinitely.
> When an element times out, the operator is supposed to emit the timeout's
> default result and stop. But that timeout result is passed through the retry
> code path, which begins with a guard meaning "a retry is already in progress
> for this
> element — ignore this completion." If a retry for the same element was just
> scheduled or is being processed at that moment, the timeout result hits that
> guard and is silently discarded. By then the timeout handler has already
> cancelled the
> element's retry timer, and the element was never marked as completed — so
> nothing remains that could ever complete it. The element is stuck permanently.
> Steps to reproduce
> {noformat}
> docker run --rm --cpus=2 -v {Path-to-flink-local-repo}:/flink -v
> {path-to-m2}:/root/.m2 -w /flink -e JAVA_HOME=/usr/lib/jvm/jdk-25.0.3+9
> apache/flink-ci-docker:java_8_11_17_21_25_maven_386_jammy ./mvnw -o -q
> -pl flink-streaming-java -Phadoop3-tests,hive3 -Djdk25 -Pjava25-target
> -Dsurefire.failIfNoSpecifiedTests=false
> -Dtest='AsyncWaitOperatorTest#testProcessingTimeWithTimeoutFunctionUnorderedWithRetry'
> surefire:test
> {noformat}
> it fails as
> {noformat}
> [ERROR]
> org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorTest.testProcessingTimeWithTimeoutFunctionUnorderedWithRetry
> -- Time elapsed: 100.1 s <<< ERROR!
> java.util.concurrent.TimeoutException:
> testProcessingTimeWithTimeoutFunctionUnorderedWithRetry() timed out after 100
> seconds
> at
> java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:511)
> at
> java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1450)
> at
> java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:2019)
> at
> java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:187)
> Suppressed: java.lang.InterruptedException: sleep interrupted
> at java.base/java.lang.Thread.sleepNanos0(Native Method)
> at java.base/java.lang.Thread.sleepNanos(Thread.java:509)
> at java.base/java.lang.Thread.sleep(Thread.java:540)
> at
> org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorTest.testProcessingTimeAlwaysTimeoutFunctionWithRetry(AsyncWaitOperatorTest.java:1444)
> at
> org.apache.flink.streaming.api.operators.async.AsyncWaitOperatorTest.testProcessingTimeWithTimeoutFunctionUnorderedWithRetry(AsyncWaitOperatorTest.java:1399)
> at java.base/java.lang.reflect.Method.invoke(Method.java:565)
> ... 4 more
> {noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
