Status

...

...

...

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

In highly-available setups with multiple masters, we need a component that provides the newly elected leader with the current state of a job that it’s trying to recover. We already have the first version of this component, the RunningJobsRegistry (RJR), that has several limitations we want to address in this FLIP.

...

• Any standby Dispatcher would know about sucessfully finished jobs which do not need to be re-executed after a failover while the job was already terminated but the JobGraph was still around (

  Jira
  server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key FLINK-11813

  ).
• The RJR does not provide access to the JobResult of a completed job. We have to return UNKNOWN as a result when we failover in application mode (

  Jira
  server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key FLINK-21928

  ).
• Having access to a JobResult, after the job has completed, would also pave the road for supporting multi-stage jobs in ApplicationMode and highly available job drivers in general.

Public Interfaces

• RunningJobsRegistry will be replaced by JobResultStore
• k8s- and ZK-specific implementations of RunningJobsRegistry will be replaced by a file-based approach
  • Users that use a customized HA implementation might be affected by this change because the HAServices interface is going to be modified
• New configuration parameters are going to be introduced to make the file-based JobResultStore implementation configurable
• REST API for the JobResult might be extended to also include the cleanup state of this Job

Proposed Changes

In this document we propose removing RJR and introducing a new opt-in component, the JobResultStore (JSR).

JobId semantics

For the purpose of this FLIP and the future work on the ApplicationMode, we need to re-define the JobId semantics to meet the following properties.

...

We don’t need any changes to interfaces around JobID as we can simply use ClusterId as a JobResultStore namespace.

JobResultStore

JobResultStore (JRS) is the successor to RJR, which is able to outlive a concrete job in order to solve recovery issues in multi-master setups. The idea of the JobResultStore is to persist information about successfully completed jobs that is under the control of a 3rd party. The cleanup of the information stored in the JRS, is the responsibility of the 3rd party (e.g. after the Flink cluster has been fully stopped).

...

• Access to the JobResult of a globally terminated job (especially useful for highly-available job drivers).
• JobResult is initially marked as dirty and needs to be committed after resource cleanup.
• Ensure that we don’t restart jobs that have already been completed.

Atomic cleanup of job resources

In multi-master setups, we need to cleanup following resources after the job finishes:

...

For a client accessing the job result we could think of providing the actual job's result since the actual computation is done even with the job still being in the cleanup phase.

Recovery workflow

Cleanup (Draft)

Currently, the different artifacts are cleaned up in different locations:

...

The plan is to provide a common interface that enables all components to trigger the cleanup based on the JobID. This cleanup component can encapsulate the retry mechanism. The actual cleanup can then be triggered either in the JobMaster or the Dispatcher depending on whether a JobGraph is still available or not.

Implementation

Interface

JobResultEntry functions as the struct to hold the actual JobResult and the corresponding cleanup state (i.e. either dirty or cleaned up). The JobResultStore interface provides all methods required to manage the JobResultEntries.

InMemoryJobResultStore (aka Standalone- or EmbeddedJobResultStore)

The InMemoryJobResultStore is a simple implementation of the JobResultStore interface that stores the data in memory (analogous to the current StandaloneRunningJobsRigstry). Any failover would result in the information being lost. This implementation can work as the default option for non-HA setups.

This implementation only covers the execution of multiple jobs without failover.

FileSystemJobResultStore

A simple persistent JobResultStore that uses a distributed file system to store the JobResultEntries in the form of files. This also provides an easy and intuitive way for the 3rd party to clean up the JRS after the Flink cluster has been terminated.

...

Used file format is versioned to allow for future extensions. Having a version present in the filename helps us to choose the right serializer / deserializer and allows for effective filtering and migration from older formats. Each serializer is responsible for writing the version inside serialized content as a safeguard against external changes. It's still an open for discussion whether we want to keep the version in the JSON structure only to avoid redundancy. This would solve the issue of conflict resolution if the content of the file does not match the filename anymore.

Configuration

The FileSystemJobResultStore would introduce two new configuration parameters:

...

This change removes a “semi-public” RunningJobsRegistry. This only affects users with custom implementations of HA Services. For regular users that are using Zookeeper or Kubernetes implementation of HA Services (which are provided by Flink), this won’t be noticeable.

Test Plan

The proposed change should be tested with unit, integration and e2e tests for common failover scenarios

Rejected Alternatives

Other Implementations of JobResultStore

Other HA implementations like KubernetesJobResultStore or ZooKeeperJobResultStore were not considered due to the following reasons:

...