[FLINK-21928] DuplicateJobSubmissionException after JobManager failover

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.10.3, 1.11.3, 1.12.2, 1.13.0
• Fix Version/s: 1.14.0
• Component/s: Runtime / Coordination
• Labels: pull-request-available stale-critical

Environment

StandaloneApplicationClusterEntryPoint using a fixed job ID, High Availability enabled

Release Note

The fix for this problem only works if the ApplicationMode is used with a single job submission and if the user code does not access the `JobExecutionResult`. If any of these conditions is violated, then Flink cannot guarantee that the whole Flink application is executed.

Additionally, it is still required that the user cleans up the corresponding HA entries for the running jobs registry because these entries won't be reliably cleaned up when encountering the situation described by FLINK-21928.

• Assignee: David Morávek
• Reporter: Ufuk Celebi
• Created: 23/Mar/21 08:37
• Updated: 28/Feb/22 12:48
• Resolved: 15/Jul/21 09:28

Description

Consider the following scenario:

• Environment: StandaloneApplicationClusterEntryPoint using a fixed job ID, high availability enabled
• Flink job reaches a globally terminal state

• Flink job is marked as finished in the high-availability service's RunningJobsRegistry

• The JobManager fails over

On recovery, the Dispatcher throws DuplicateJobSubmissionException, because the job is marked as done in the RunningJobsRegistry.

When this happens, users cannot get out of the situation without manually redeploying the JobManager process and changing the job ID^1^.

The desired semantics are that we don't want to re-execute a job that has reached a globally terminal state. In this particular case, we know that the job has already reached such a state (as it has been marked in the registry). Therefore, we could handle this case by executing the regular termination sequence instead of throwing a DuplicateJobSubmission.

—

¹ With ZooKeeper HA, the respective node is not ephemeral. In Kubernetes HA, there is no  notion of ephemeral data that is tied to a session in the first place afaik.

 

Issue Links

• is duplicated by: FLINK-26333 Repeated exception with DuplicateJobSubmissionException: Job has already been submitted
• is related to: FLINK-11813 Standby per job mode Dispatchers don't know job's JobSchedulingStatus
• relates to: FLINK-21979 Job can be restarted from the beginning after it reached a terminal state
• relates to: FLINK-21980 ZooKeeperRunningJobsRegistry creates an empty znode
• links to: GitHub Pull Request #16464