Mohsen Rezaei created FLINK-41354:
-------------------------------------
Summary: Dispatcher never terminates when it is stopped while a
cleanup retry is pending, so the next leader never starts
Key: FLINK-41354
URL: https://issues.apache.org/jira/browse/FLINK-41354
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Affects Versions: 2.1.3, 2.2.1, 2.3.0, 2.0.2, 2.4.1
Reporter: Mohsen Rezaei
If the Dispatcher is stopped while the cleanup of a globally terminated job is
waiting for a retry, it never finishes stopping. With Kubernetes HA this leaves
the JobManager holding the leader lease while its REST API answers {{HTTP 503:
Service temporarily unavailable
due to an ongoing leader election}} until the process is restarted, because
{{DefaultDispatcherRunner}} only starts the next {{DispatcherLeaderProcess}}
after the previous one has terminated.
h3. Observed
Flink 2.3.0, standalone session cluster, Kubernetes HA, one JobManager:
# A job reaches FAILED ({{restart-strategy: none}}; FINISHED and CANCELED
behave the same).
# Its first global cleanup attempt fails and a retry is scheduled. I caused the
failure by changing the holder in the leader {{ConfigMap}}: the execution plan
store only writes while the JobManager is the leader ({{Could not remove
execution plan with job id ... from KubernetesStateHandleStore}}).
# About 370 ms later: {{DefaultDispatcherRunner was revoked the leadership}}.
About 8s later the JobManager regains the lease:
{{DefaultDispatcherRunner was granted leadership ... Creating new
DispatcherLeaderProcess}}.
# Nothing is logged after that, and REST answers 503 for as long as the job is
active (over 7 minutes in testing).
The same lease loss with a running job and no cleanup in flight recovers in
about 9s. A new unit test reproduces the hang on {{release-2.3.0}} and on
{{master}}. It is modelled on
{{DispatcherCleanupITCase#testCleanupAfterLeadershipChange}}, with a retrying
cleanup strategy instead of {{NONE}} and the {{Dispatcher}} stopped after the
first failed attempt.
Any failed cleanup attempt shortly before the Dispatcher stops can trigger it,
for example another JobManager taking over the lease, or the Kubernetes API
failing.
h3. Cause
* {{RpcEndpoint#internalCallOnStop}} closes the resource registry first. That
shuts down the {{MainThreadExecutor}}'s scheduled executor
(FLINK-25085) and drops every pending scheduled task.
* {{DefaultResourceCleaner}} schedules cleanup retries on that executor
({{FutureUtils#retryWithDelay}}, FLINK-25433), so a pending retry
is dropped and its future never completes.
* {{Dispatcher#onStop}} waits for {{jobManagerRunnerTerminationFutures}}, which
for a globally terminated job include that cleanup
({{Dispatcher#removeJob}}).
{{testCleanupAfterLeadershipChange}} uses cleanup strategy {{NONE}}, so this
case is not covered. The default strategy ({{exponential-delay}}) retries
without a limit.
h3. Possible fix
Let {{ResourceCleaner}} stop retrying, and call it at the start of
{{Dispatcher#onStop}}. A cleanup waiting for a retry then fails instead of
hanging, and a running attempt is not retried if it fails. An unfinished global
cleanup stays dirty in the {{JobResultStore}}, and the next leader resumes it.
I have this working with a unit test, and the
{{org.apache.flink.runtime.dispatcher}} tests pass on 2.3.0.
[~mapohl], does this direction sound right to you? If so, you can assign the
ticket to me and I'll open a PR for the fix.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
