[
https://issues.apache.org/jira/browse/FLINK-40887?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141746#comment-18141746
]
Gautham Sankar V commented on FLINK-40887:
------------------------------------------
Hi [~martijnvisser]
I looked into this failure and I believe the cause is in the test, not in the
AdaptiveScheduler.
*Cause*
{\{TaskFailureITCase}} runs the collect iterator without checkpointing, which
selects the buffer with \{{failureTolerance=false}} ("mode A"). In that mode,
if the client has already seen the sink version of attempt 0 and the restart
brings a new sink version, the iterator throws \{{RuntimeException("Job
restarted")}}. The test then fails, because it asserts that \{{Test
Exception:1}} is in the cause chain.
The mapper fails right away, so attempt 0 is reachable for only a few
milliseconds. The failure needs a client poll to land in that window. From the
failing nightly run (mvn-4.log, \{{testJobWithoutObjectReuse}}):
{code}
02:29:27,935 attempt 0 sink server is up
02:29:27,952 coordinator connects to the attempt 0 sink
02:29:27,956 client's empty-version request answered with b801fd04
02:29:27,973 attempt 0 fails (Test Exception:0)
02:29:28,027 old CollectSinkOperatorCoordinator closed
02:29:28,070 new coordinator receives attempt 1's address
02:29:28,079 version mismatch (expected b801fd04), "Job restarted"
{code}
The client holds attempt 0's version 17 ms before attempt 0 fails, so attempt
1's different version is a mismatch. A poll before the sink answers or after
attempt 0 dies leaves the client with no older version, and the test passes. I
don't know why it shows up in the AdaptiveScheduler leg only. CI load may
change how often a poll lands in the window, but nothing in the log shows that.
*Reproduction*
I could not reproduce the natural flake in a full CI run with
\{{-Penable-adaptive-scheduler}}, nor in 30 local runs of
\{{TaskFailureITCase}}. I could reproduce the same failure locally by making
the window wide: with a mapper that keeps attempt 0 alive for 1 s, the client
always sees attempt 0's version. The unchanged assertion then fails in mode A
with the same message as CI ("Expecting Optional to contain a value but it was
empty", caused by "Job restarted"), and it passes with exactly-once and
at-least-once checkpointing.
*Proposed fix*
Enable at-least-once checkpointing in \{{TaskFailureITCase}} (10 minute
interval, so no checkpoint completes). The iterator then uses the
\{{failureTolerance=true}} buffer ("mode C"): on a version mismatch it resets
and keeps reading, and the final job failure is reported whichever attempt the
client polled. The assertion stays unchanged and strict.
The modified \{{TaskFailureITCase}} passed 20/20 locally with the
AdaptiveScheduler.
I will open a PR with the test change.
> TaskFailureITCase.testJobWithoutObjectReuse fails with "Expecting Optional to
> contain a value but it was empty"
> ---------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40887
> URL: https://issues.apache.org/jira/browse/FLINK-40887
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
> Affects Versions: 2.2.2, 2.4.0
> Reporter: Martijn Visser
> Assignee: Gautham Sankar V
> Priority: Major
> Labels: test-stability
>
> https://github.com/apache/flink/actions/runs/36954250535/job/110677877340
> (master nightly 2026-10-02, AdaptiveScheduler profile)
> {code}
> [ERROR]
> org.apache.flink.test.example.failing.TaskFailureITCase.testJobWithoutObjectReuse
> -- Time elapsed: 1.338 s <<< FAILURE!
> java.lang.AssertionError:
> Expecting Optional to contain a value but it was empty.
> at
> org.apache.flink.test.example.failing.TaskFailureITCase.testProgram(TaskFailureITCase.java:57)
> at
> org.apache.flink.test.util.JavaProgramTestBase.executeProgramMultipleTimes(JavaProgramTestBase.java:108)
> at
> org.apache.flink.test.util.JavaProgramTestBase.testJobWithoutObjectReuse(JavaProgramTestBase.java:136)
> {code}
> The job itself failed as expected with "This is an expected Test Exception:1"
> after one restart, but the exception the client caught does not contain it.
> In the test log the collect client reached the sink of both attempts:
> {code}
> 02:31:35,262 [Thread-8] INFO
> org.apache.flink.streaming.api.operators.collect.CollectSinkFunction [] -
> Invalid request. Received version = , offset = 0, while expected version =
> 0cbf0890-bb27-4181-a956-cb28a988886f, offset = 0
> 02:31:35,392 [Thread-10] INFO
> org.apache.flink.streaming.api.operators.collect.CollectSinkFunction [] -
> Invalid request. Received version = 0cbf0890-bb27-4181-a956-cb28a988886f,
> offset = 0, while expected version = 672ae5ae-b048-4a8a-a231-979d262d8eb1,
> offset = 0
> {code}
> In the passing testJobWithObjectReuse run right after it, it reached neither.
> The release-2.2 failure on 2026-09-11 has the same two lines.
> It failed before in the AdaptiveScheduler tests leg of these nightlies, never
> in the other profiles:
> https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=71463&view=logs&j=8fd9202e-fd17-5b26-353c-ac1ff76c8f28
> (master, 2025-12-31)
> https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=76172&view=logs&j=8fd9202e-fd17-5b26-353c-ac1ff76c8f28
> (release-2.2, 2026-06-19)
> https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=77101&view=logs&j=8fd9202e-fd17-5b26-353c-ac1ff76c8f28
> (master, 2026-07-15)
> https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=78966&view=logs&j=8fd9202e-fd17-5b26-353c-ac1ff76c8f28
> (release-2.2, 2026-09-11)
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
