org.apache.flink.runtime.clusterframework

Class FlinkResourceManager<WorkerType extends ResourceIDRetrievable>

java.lang.Object

akka.actor.UntypedActor

org.apache.flink.runtime.akka.FlinkUntypedActor

org.apache.flink.runtime.clusterframework.FlinkResourceManager<WorkerType>

All Implemented Interfaces:

akka.actor.Actor

Direct Known Subclasses:

MesosFlinkResourceManager, StandaloneResourceManager, YarnFlinkResourceManager

public abstract class FlinkResourceManager<WorkerType extends ResourceIDRetrievable>
extends FlinkUntypedActor

Worker allocation steps

1. The resource manager decides to request more workers. This can happen in order to fill the initial pool, or as a result of the JobManager requesting more workers.
2. The resource master calls requestNewWorkers(int), which triggers requests for more containers. After that, the getNumWorkerRequestsPending() should reflect the pending requests.
3. The concrete framework may acquire containers and then trigger to start TaskManagers in those containers. That should be reflected in getNumWorkersPendingRegistration().
4. At some point, the TaskManager processes will have started and send a registration message to the JobManager. The JobManager will perform a lookup with the ResourceManager to check if it really started this TaskManager. The method workerStarted(ResourceID) will be called to inform about a registered worker.

Nested Class Summary

Nested classes/interfaces inherited from interface akka.actor.Actor

akka.actor.Actor.emptyBehavior$, akka.actor.Actor.ignoringBehavior$

Field Summary

Fields

Modifier and Type	Field and Description
protected Configuration	config The Flink configuration object
protected static int	EXIT_CODE_FATAL_ERROR The exit code with which the process is stopped in case of a fatal error
static String	RESOURCE_MANAGER_NAME The default name of the resource manager actor

Fields inherited from class org.apache.flink.runtime.akka.FlinkUntypedActor

LOG

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	FlinkResourceManager(int numInitialTaskManagers, Configuration flinkConfig, LeaderRetrievalService leaderRetriever) Creates a AbstractFrameworkMaster actor.

Method Summary

Modifier and Type	Method and Description
Collection<WorkerType>	allStartedWorkers() Gets an iterable for all currently started TaskManagers.
protected abstract void	fatalError(String message, Throwable error) Notifies the resource master of a fatal error.
int	getDesignatedWorkerPoolSize() Gets the current designated worker pool size, meaning the number of workers that the resource master strives to maintain.
protected UUID	getLeaderSessionID() Returns the current leader session ID associated with this actor.
int	getNumberOfStartedTaskManagers() Gets the number of currently started TaskManagers.
protected abstract int	getNumWorkerRequestsPending() Gets the number of requested workers that have not yet been granted.
protected abstract int	getNumWorkersPendingRegistration() Gets the number of containers that have been started, but where the TaskManager has not yet registered at the job manager.
static akka.actor.Props	getResourceManagerProps(Class<? extends FlinkResourceManager> resourceManagerClass, Configuration configuration, LeaderRetrievalService leaderRetrievalService)
Collection<WorkerType>	getStartedTaskManagers() Gets the currently registered resources.
protected void	handleMessage(Object message) This method receives the actor messages after they have been filtered for a match with the leader session.
protected abstract void	initialize() Initializes the framework specific components.
boolean	isStarted(ResourceID resourceId) Gets the started worker for a given resource ID, if one is available.
void	notifyWorkerFailed(ResourceID resourceID, String message) This method should be called by the framework once it detects that a currently registered worker has failed.
void	postStop()
void	preStart()
protected abstract Collection<WorkerType>	reacceptRegisteredWorkers(Collection<ResourceID> registered) This method is called when the resource manager starts after a failure and reconnects to the leader JobManager, who still has some workers registered.
protected abstract void	releasePendingWorker(ResourceID resourceID) Trigger a release of a pending worker.
protected abstract void	releaseStartedWorker(WorkerType resourceID) Trigger a release of a started worker.
protected abstract void	requestNewWorkers(int numWorkers) Requests to allocate a certain number of new workers.
protected void	sendInfoMessage(String message)
protected abstract void	shutdownApplication(ApplicationStatus finalStatus, String optionalDiagnostics) The framework specific code for shutting down the application.
static akka.actor.ActorRef	startResourceManagerActors(Configuration configuration, akka.actor.ActorSystem actorSystem, LeaderRetrievalService leaderRetriever, Class<? extends FlinkResourceManager<?>> resourceManagerClass) Starts the resource manager actors.
static akka.actor.ActorRef	startResourceManagerActors(Configuration configuration, akka.actor.ActorSystem actorSystem, LeaderRetrievalService leaderRetriever, Class<? extends FlinkResourceManager<?>> resourceManagerClass, String resourceManagerActorName) Starts the resource manager actors.
void	triggerCheckWorkers() This method causes the resource framework master to asynchronouslyre-examine the set of available and pending workers containers, and release or allocate containers if needed.
protected void	triggerConnectingToJobManager(String leaderAddress) Causes the resource manager to announce itself at the new leader JobManager and obtains its connection information and currently known TaskManagers.
protected abstract WorkerType	workerStarted(ResourceID resourceID) Callback when a worker was started.

Methods inherited from class org.apache.flink.runtime.akka.FlinkUntypedActor

decorateMessage, onReceive

Methods inherited from class akka.actor.UntypedActor

akka$actor$Actor$_setter_$context_$eq, akka$actor$Actor$_setter_$self_$eq, aroundPostRestart, aroundPostStop, aroundPreRestart, aroundPreStart, aroundReceive, context, getContext, getSelf, getSender, postRestart, preRestart, receive, self, sender, supervisorStrategy, unhandled

Methods inherited from class java.lang.Object

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

Field Detail

EXIT_CODE_FATAL_ERROR

protected static final int EXIT_CODE_FATAL_ERROR

The exit code with which the process is stopped in case of a fatal error

See Also:

Constant Field Values

RESOURCE_MANAGER_NAME

public static final String RESOURCE_MANAGER_NAME

The default name of the resource manager actor

See Also:

Constant Field Values

config

protected final Configuration config

The Flink configuration object

Constructor Detail

FlinkResourceManager

protected FlinkResourceManager(int numInitialTaskManagers,
                               Configuration flinkConfig,
                               LeaderRetrievalService leaderRetriever)

Creates a AbstractFrameworkMaster actor.

Parameters:

flinkConfig - The Flink configuration object.

Method Detail

preStart

public void preStart()

Specified by:

preStart in interface akka.actor.Actor

Overrides:

preStart in class akka.actor.UntypedActor

postStop

public void postStop()

Specified by:

postStop in interface akka.actor.Actor

Overrides:

postStop in class akka.actor.UntypedActor

handleMessage

protected void handleMessage(Object message)

This method receives the actor messages after they have been filtered for a match with the leader session.

Specified by:

handleMessage in class FlinkUntypedActor

Parameters:

message - The incoming actor message.

getLeaderSessionID

protected final UUID getLeaderSessionID()

Description copied from class: FlinkUntypedActor

Returns the current leader session ID associated with this actor.

Specified by:

getLeaderSessionID in class FlinkUntypedActor

Returns:

getDesignatedWorkerPoolSize

public int getDesignatedWorkerPoolSize()

Gets the current designated worker pool size, meaning the number of workers that the resource master strives to maintain. The actual number of workers may be lower (if worker requests are still pending) or higher (if workers have not yet been released).

Returns:

The designated worker pool size.

getNumberOfStartedTaskManagers

public int getNumberOfStartedTaskManagers()

Gets the number of currently started TaskManagers.

Returns:

The number of currently started TaskManagers.

getStartedTaskManagers

public Collection<WorkerType> getStartedTaskManagers()

Gets the currently registered resources.

Returns:

isStarted

public boolean isStarted(ResourceID resourceId)

Gets the started worker for a given resource ID, if one is available.

Parameters:

resourceId - The resource ID for the worker.

Returns:

True if already registered, otherwise false

allStartedWorkers

public Collection<WorkerType> allStartedWorkers()

Gets an iterable for all currently started TaskManagers.

Returns:

All currently started TaskManagers.

triggerConnectingToJobManager

protected void triggerConnectingToJobManager(String leaderAddress)

Causes the resource manager to announce itself at the new leader JobManager and obtains its connection information and currently known TaskManagers.

Parameters:

leaderAddress - The akka actor URL of the new leader JobManager.

triggerCheckWorkers

public void triggerCheckWorkers()

This method causes the resource framework master to asynchronouslyre-examine the set of available and pending workers containers, and release or allocate containers if needed. The method sends an actor message which will trigger the re-examination.

notifyWorkerFailed

public void notifyWorkerFailed(ResourceID resourceID,
                               String message)

This method should be called by the framework once it detects that a currently registered worker has failed.

Parameters:

resourceID - Id of the worker that has failed.

message - An informational message that explains why the worker failed.

initialize

protected abstract void initialize()
                            throws Exception

Initializes the framework specific components.

Throws:

Exception - Exceptions during initialization cause the resource manager to fail. If the framework is able to recover this resource manager, it will be restarted.

shutdownApplication

protected abstract void shutdownApplication(ApplicationStatus finalStatus,
                                            String optionalDiagnostics)

The framework specific code for shutting down the application. This should report the application's final status and shut down the resource manager cleanly. This method also needs to make sure all pending containers that are not registered yet are returned.

Parameters:

finalStatus - The application status to report.

optionalDiagnostics - An optional diagnostics message.

fatalError

protected abstract void fatalError(String message,
                                   Throwable error)

Notifies the resource master of a fatal error.

IMPORTANT: This should not cleanly shut down this master, but exit it in such a way that a high-availability setting would restart this or fail over to another master.

requestNewWorkers

protected abstract void requestNewWorkers(int numWorkers)

Requests to allocate a certain number of new workers.

Parameters:

numWorkers - The number of workers to allocate.

releasePendingWorker

protected abstract void releasePendingWorker(ResourceID resourceID)

Trigger a release of a pending worker.

Parameters:

resourceID - The worker resource id

releaseStartedWorker

protected abstract void releaseStartedWorker(WorkerType resourceID)

Trigger a release of a started worker.

Parameters:

resourceID - The worker resource id

workerStarted

protected abstract WorkerType workerStarted(ResourceID resourceID)

Callback when a worker was started.

Parameters:

resourceID - The worker resource id

reacceptRegisteredWorkers

protected abstract Collection<WorkerType> reacceptRegisteredWorkers(Collection<ResourceID> registered)

This method is called when the resource manager starts after a failure and reconnects to the leader JobManager, who still has some workers registered. The method is used to consolidate the view between resource manager and JobManager. The resource manager gets the list of TaskManagers that the JobManager considers available and should return a list or nodes that the resource manager considers available. After that, the JobManager is informed of loss of all TaskManagers that are not part of the returned list. It is possible that the resource manager initially confirms some TaskManagers to be alive, even through they are in an uncertain status, if it later sends necessary failure notifications via calling notifyWorkerFailed(ResourceID, String).

Parameters:

registered - The list of TaskManagers that the JobManager knows.

Returns:

The subset of TaskManagers that the resource manager can confirm to be alive.

getNumWorkerRequestsPending

protected abstract int getNumWorkerRequestsPending()

Gets the number of requested workers that have not yet been granted.

Returns:

The number pending worker requests. Must never be smaller than 0.

getNumWorkersPendingRegistration

protected abstract int getNumWorkersPendingRegistration()

Gets the number of containers that have been started, but where the TaskManager has not yet registered at the job manager.

Returns:

The number of started containers pending TaskManager registration. Must never be smaller than 0.

sendInfoMessage

protected void sendInfoMessage(String message)

startResourceManagerActors

public static akka.actor.ActorRef startResourceManagerActors(Configuration configuration,
                                                             akka.actor.ActorSystem actorSystem,
                                                             LeaderRetrievalService leaderRetriever,
                                                             Class<? extends FlinkResourceManager<?>> resourceManagerClass)

Starts the resource manager actors.

Parameters:

configuration - The configuration for the resource manager

actorSystem - The actor system to start the resource manager in

leaderRetriever - The leader retriever service to initialize the resource manager

resourceManagerClass - The class of the ResourceManager to be started

Returns:

ActorRef of the resource manager

startResourceManagerActors

public static akka.actor.ActorRef startResourceManagerActors(Configuration configuration,
                                                             akka.actor.ActorSystem actorSystem,
                                                             LeaderRetrievalService leaderRetriever,
                                                             Class<? extends FlinkResourceManager<?>> resourceManagerClass,
                                                             String resourceManagerActorName)

Starts the resource manager actors.

Parameters:

configuration - The configuration for the resource manager

actorSystem - The actor system to start the resource manager in

leaderRetriever - The leader retriever service to initialize the resource manager

resourceManagerClass - The class of the ResourceManager to be started

resourceManagerActorName - The name of the resource manager actor.

Returns:

ActorRef of the resource manager

getResourceManagerProps

public static akka.actor.Props getResourceManagerProps(Class<? extends FlinkResourceManager> resourceManagerClass,
                                                       Configuration configuration,
                                                       LeaderRetrievalService leaderRetrievalService)