org.apache.flink.yarn.highavailability

Class YarnHighAvailabilityServices

java.lang.Object

org.apache.flink.yarn.highavailability.YarnHighAvailabilityServices

All Implemented Interfaces:

AutoCloseable, org.apache.flink.runtime.highavailability.HighAvailabilityServices

Direct Known Subclasses:

AbstractYarnNonHaServices

public abstract class YarnHighAvailabilityServices
extends Object
implements org.apache.flink.runtime.highavailability.HighAvailabilityServices

The basis of HighAvailabilityServices for YARN setups. These high-availability services auto-configure YARN's HDFS and the YARN application's working directory to be used to store job recovery data.

Note for implementers: This class locks access to and creation of services, to make sure all services are properly shut down when shutting down this class. To participate in the checks, overriding methods should frame method body with calls to enter() and exit() as shown in the following pattern:

 public LeaderRetrievalService getResourceManagerLeaderRetriever() {
     enter();
     try {
         CuratorClient client = getCuratorClient();
         return ZooKeeperUtils.createLeaderRetrievalService(client, configuration, RESOURCE_MANAGER_LEADER_PATH);
     } finally {
         exit();
     }
 }
 

Field Summary

Fields

Modifier and Type	Field and Description
protected org.apache.flink.runtime.blob.BlobStoreService	blobStoreService Blob store service to be used for the BlobServer and BlobCache.
static String	FLINK_RECOVERY_DATA_DIR The name of the sub directory in which Flink stores the recovery data.
protected org.apache.flink.core.fs.FileSystem	flinkFileSystem The Flink FileSystem object that represent the HDFS used by YARN.
protected org.apache.flink.core.fs.Path	haDataDirectory The directory for HA persistent data.
protected org.apache.hadoop.fs.FileSystem	hadoopFileSystem The Hadoop FileSystem object that represent the HDFS used by YARN.
protected static org.slf4j.Logger	LOG Logger for these services, shared with subclasses.
protected org.apache.flink.core.fs.Path	workingDirectory The working directory of this YARN application.

Fields inherited from interface org.apache.flink.runtime.highavailability.HighAvailabilityServices

DEFAULT_JOB_ID, DEFAULT_LEADER_ID

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	YarnHighAvailabilityServices(org.apache.flink.configuration.Configuration config, org.apache.hadoop.conf.Configuration hadoopConf) Creates new YARN high-availability services, configuring the file system and recovery data directory based on the working directory in the given Hadoop configuration.

Method Summary

Modifier and Type	Method and Description
void	close()
void	closeAndCleanupAllData()
org.apache.flink.runtime.blob.BlobStore	createBlobStore()
static YarnHighAvailabilityServices	forSingleJobAppMaster(org.apache.flink.configuration.Configuration flinkConfig, org.apache.hadoop.conf.Configuration hadoopConfig) Creates the high-availability services for a single-job Flink YARN application, to be used in the Application Master that runs both ResourceManager and JobManager.
static YarnHighAvailabilityServices	forYarnTaskManager(org.apache.flink.configuration.Configuration flinkConfig, org.apache.hadoop.conf.Configuration hadoopConfig) Creates the high-availability services for the TaskManagers participating in a Flink YARN application.
boolean	isClosed() Checks whether these services have been shut down.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.flink.runtime.highavailability.HighAvailabilityServices

getCheckpointRecoveryFactory, getDispatcherLeaderElectionService, getDispatcherLeaderRetriever, getJobManagerLeaderElectionService, getJobManagerLeaderRetriever, getJobManagerLeaderRetriever, getResourceManagerLeaderElectionService, getResourceManagerLeaderRetriever, getRunningJobsRegistry, getSubmittedJobGraphStore, getWebMonitorLeaderElectionService, getWebMonitorLeaderRetriever

Field Detail

FLINK_RECOVERY_DATA_DIR

public static final String FLINK_RECOVERY_DATA_DIR

The name of the sub directory in which Flink stores the recovery data.

See Also:

Constant Field Values

LOG

protected static final org.slf4j.Logger LOG

Logger for these services, shared with subclasses.

flinkFileSystem

protected final org.apache.flink.core.fs.FileSystem flinkFileSystem

The Flink FileSystem object that represent the HDFS used by YARN.

hadoopFileSystem

protected final org.apache.hadoop.fs.FileSystem hadoopFileSystem

The Hadoop FileSystem object that represent the HDFS used by YARN.

workingDirectory

protected final org.apache.flink.core.fs.Path workingDirectory

The working directory of this YARN application. This MUST NOT be deleted when the HA services clean up

haDataDirectory

protected final org.apache.flink.core.fs.Path haDataDirectory

The directory for HA persistent data. This should be deleted when the HA services clean up.

blobStoreService

protected final org.apache.flink.runtime.blob.BlobStoreService blobStoreService

Blob store service to be used for the BlobServer and BlobCache.

Constructor Detail

YarnHighAvailabilityServices

protected YarnHighAvailabilityServices(org.apache.flink.configuration.Configuration config,
                                       org.apache.hadoop.conf.Configuration hadoopConf)
                                throws IOException

Creates new YARN high-availability services, configuring the file system and recovery data directory based on the working directory in the given Hadoop configuration.

This class requires that the default Hadoop file system configured in the given Hadoop configuration is an HDFS.

Parameters:

config - The Flink configuration of this component / process.

hadoopConf - The Hadoop configuration for the YARN cluster.

Throws:

IOException - Thrown, if the initialization of the Hadoop file system used by YARN fails.

Method Detail

createBlobStore

public org.apache.flink.runtime.blob.BlobStore createBlobStore()
                                                        throws IOException

Specified by:

createBlobStore in interface org.apache.flink.runtime.highavailability.HighAvailabilityServices

Throws:

IOException

isClosed

public boolean isClosed()

Checks whether these services have been shut down.

Returns:

True, if this instance has been shut down, false if it still operational.

close

public void close()
           throws Exception

Specified by:

close in interface AutoCloseable

Specified by:

close in interface org.apache.flink.runtime.highavailability.HighAvailabilityServices

Throws:

Exception

closeAndCleanupAllData

public void closeAndCleanupAllData()
                            throws Exception

Specified by:

closeAndCleanupAllData in interface org.apache.flink.runtime.highavailability.HighAvailabilityServices

Throws:

Exception

forSingleJobAppMaster

public static YarnHighAvailabilityServices forSingleJobAppMaster(org.apache.flink.configuration.Configuration flinkConfig,
                                                                 org.apache.hadoop.conf.Configuration hadoopConfig)
                                                          throws IOException

Creates the high-availability services for a single-job Flink YARN application, to be used in the Application Master that runs both ResourceManager and JobManager.

Parameters:

flinkConfig - The Flink configuration.

hadoopConfig - The Hadoop configuration for the YARN cluster.

Returns:

The created high-availability services.

Throws:

IOException - Thrown, if the high-availability services could not be initialized.

forYarnTaskManager

public static YarnHighAvailabilityServices forYarnTaskManager(org.apache.flink.configuration.Configuration flinkConfig,
                                                              org.apache.hadoop.conf.Configuration hadoopConfig)
                                                       throws IOException

Creates the high-availability services for the TaskManagers participating in a Flink YARN application.

Parameters:

flinkConfig - The Flink configuration.

hadoopConfig - The Hadoop configuration for the YARN cluster.

Returns:

The created high-availability services.

Throws:

IOException - Thrown, if the high-availability services could not be initialized.