org.apache.flink.runtime.jobmanager

Class JobManager$

java.lang.Object

org.apache.flink.runtime.jobmanager.JobManager$

public class JobManager$
extends Object

Job Manager companion object. Contains the entry point (main method) to run the JobManager in a standalone fashion. Also contains various utility methods to start the JobManager and to look up the JobManager actor reference.

Field Summary

Fields

Modifier and Type	Field and Description
static JobManager$	MODULE$ Static reference to the singleton instance of this Scala object.

Constructor Summary

Constructors

Constructor and Description
JobManager$()

Method Summary

Modifier and Type	Method and Description
String	ARCHIVE_NAME() Name of the archive actor
scala.Tuple12<ExecutorService,InstanceManager,Scheduler,BlobLibraryCacheManager,RestartStrategy,scala.concurrent.duration.FiniteDuration,Object,LeaderElectionService,SubmittedJobGraphStore,CheckpointRecoveryFactory,SavepointStore,scala.concurrent.duration.FiniteDuration>	createJobManagerComponents(Configuration configuration, scala.Option<LeaderElectionService> leaderElectionServiceOption) Create the job manager components as (instanceManager, scheduler, libraryCacheManager, archiverProps, defaultExecutionRetries, delayBetweenRetries, timeout)
akka.actor.ActorRef	getJobManagerActorRef(InetSocketAddress address, akka.actor.ActorSystem system, Configuration config) Resolves the JobManager actor reference in a blocking fashion.
akka.actor.ActorRef	getJobManagerActorRef(InetSocketAddress address, akka.actor.ActorSystem system, scala.concurrent.duration.FiniteDuration timeout) Resolves the JobManager actor reference in a blocking fashion.
akka.actor.ActorRef	getJobManagerActorRef(String jobManagerUrl, akka.actor.ActorSystem system, scala.concurrent.duration.FiniteDuration timeout) Resolves the JobManager actor reference in a blocking fashion.
scala.concurrent.Future<akka.actor.ActorRef>	getJobManagerActorRefFuture(InetSocketAddress address, akka.actor.ActorSystem system, scala.concurrent.duration.FiniteDuration timeout)
String	getJobManagerAkkaURL(akka.actor.ActorSystem system, scala.Option<String> name)
String	getLocalJobManagerAkkaURL(scala.Option<String> name) Builds the akka actor path for the JobManager actor to address the actor within its own actor system.
String	getRemoteJobManagerAkkaURL(Configuration config) Returns the JobManager actor's remote Akka URL, given the configured hostname and port.
String	getRemoteJobManagerAkkaURL(InetSocketAddress address, scala.Option<String> name) Builds the akka actor path for the JobManager actor, given the socket address where the JobManager's actor system runs.
String	JOB_MANAGER_NAME() Name of the JobManager actor
grizzled.slf4j.Logger	LOG()
void	main(String[] args) Entry point (main method) to run the JobManager in a standalone fashion.
scala.Tuple4<Configuration,JobManagerMode,String,Iterator<Integer>>	parseArgs(String[] args) Loads the configuration, execution mode and the listening address from the provided command line arguments.
<T> scala.util.Try<T>	retryOnBindException(scala.Function0<T> fn, scala.Function0<Object> stopCond, long maxSleepBetweenRetries) Retries a function if it fails because of a BindException.
void	runJobManager(Configuration configuration, JobManagerMode executionMode, String listeningAddress, int listeningPort) Starts and runs the JobManager with all its components.
void	runJobManager(Configuration configuration, JobManagerMode executionMode, String listeningAddress, Iterator<Integer> listeningPortRange) Starts and runs the JobManager with all its components trying to bind to a port in the specified range.
int	RUNTIME_FAILURE_RETURN_CODE()
akka.actor.ActorRef	startActor(akka.actor.Props props, akka.actor.ActorSystem actorSystem)
scala.Tuple4<akka.actor.ActorSystem,akka.actor.ActorRef,akka.actor.ActorRef,scala.Option<WebMonitor>>	startActorSystemAndJobManagerActors(Configuration configuration, JobManagerMode executionMode, String listeningAddress, int listeningPort, Class<? extends JobManager> jobManagerClass, Class<? extends MemoryArchivist> archiveClass) Starts an ActorSystem, the JobManager and all its components including the WebMonitor.
scala.Tuple2<akka.actor.ActorRef,akka.actor.ActorRef>	startJobManagerActors(Configuration configuration, akka.actor.ActorSystem actorSystem, Class<? extends JobManager> jobManagerClass, Class<? extends MemoryArchivist> archiveClass) Starts the JobManager and job archiver based on the given configuration, in the given actor system.
scala.Tuple2<akka.actor.ActorRef,akka.actor.ActorRef>	startJobManagerActors(Configuration configuration, akka.actor.ActorSystem actorSystem, scala.Option<String> jobMangerActorName, scala.Option<String> archiveActorName, Class<? extends JobManager> jobManagerClass, Class<? extends MemoryArchivist> archiveClass) Starts the JobManager and job archiver based on the given configuration, in the given actor system.
int	STARTUP_FAILURE_RETURN_CODE()

Methods inherited from class java.lang.Object

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

Field Detail

MODULE$

public static final JobManager$ MODULE$

Static reference to the singleton instance of this Scala object.

Constructor Detail

JobManager$

public JobManager$()

Method Detail

LOG

public grizzled.slf4j.Logger LOG()

STARTUP_FAILURE_RETURN_CODE

public int STARTUP_FAILURE_RETURN_CODE()

RUNTIME_FAILURE_RETURN_CODE

public int RUNTIME_FAILURE_RETURN_CODE()

JOB_MANAGER_NAME

public String JOB_MANAGER_NAME()

Name of the JobManager actor

ARCHIVE_NAME

public String ARCHIVE_NAME()

Name of the archive actor

main

public void main(String[] args)

Entry point (main method) to run the JobManager in a standalone fashion.

Parameters:

args - The command line arguments.

runJobManager

public void runJobManager(Configuration configuration,
                          JobManagerMode executionMode,
                          String listeningAddress,
                          int listeningPort)

Starts and runs the JobManager with all its components. First, this method starts a dedicated actor system for the JobManager. Second, its starts all components of the JobManager (including library cache, instance manager, scheduler). Finally, it starts the JobManager actor itself.

This method blocks indefinitely (or until the JobManager's actor system is shut down).

Parameters:

configuration - The configuration object for the JobManager.

executionMode - The execution mode in which to run. Execution mode LOCAL will spawn an an additional TaskManager in the same process.

listeningAddress - The hostname where the JobManager should listen for messages.

listeningPort - The port where the JobManager should listen for messages.

runJobManager

public void runJobManager(Configuration configuration,
                          JobManagerMode executionMode,
                          String listeningAddress,
                          Iterator<Integer> listeningPortRange)

Starts and runs the JobManager with all its components trying to bind to a port in the specified range.

Parameters:

configuration - The configuration object for the JobManager.

executionMode - The execution mode in which to run. Execution mode LOCAL will spawn an an additional TaskManager in the same process.

listeningAddress - The hostname where the JobManager should listen for messages.

listeningPortRange - The port range where the JobManager should listen for messages.

retryOnBindException

public <T> scala.util.Try<T> retryOnBindException(scala.Function0<T> fn,
                                                  scala.Function0<Object> stopCond,
                                                  long maxSleepBetweenRetries)

Retries a function if it fails because of a BindException.

Parameters:

fn - The function to retry

stopCond - Flag to signal termination

maxSleepBetweenRetries - Max random sleep time between retries

Returns:

Return value of the the function to retry

startActorSystemAndJobManagerActors

public scala.Tuple4<akka.actor.ActorSystem,akka.actor.ActorRef,akka.actor.ActorRef,scala.Option<WebMonitor>> startActorSystemAndJobManagerActors(Configuration configuration,
                                                                                                                                                 JobManagerMode executionMode,
                                                                                                                                                 String listeningAddress,
                                                                                                                                                 int listeningPort,
                                                                                                                                                 Class<? extends JobManager> jobManagerClass,
                                                                                                                                                 Class<? extends MemoryArchivist> archiveClass)

Starts an ActorSystem, the JobManager and all its components including the WebMonitor.

Parameters:

configuration - The configuration object for the JobManager

executionMode - The execution mode in which to run. Execution mode LOCAL with spawn an additional TaskManager in the same process.

listeningAddress - The hostname where the JobManager should listen for messages.

listeningPort - The port where the JobManager should listen for messages

jobManagerClass - The class of the JobManager to be started

archiveClass - The class of the Archivist to be started

Returns:

A tuple containing the started ActorSystem, ActorRefs to the JobManager and the Archivist and an Option containing a possibly started WebMonitor

parseArgs

public scala.Tuple4<Configuration,JobManagerMode,String,Iterator<Integer>> parseArgs(String[] args)

Loads the configuration, execution mode and the listening address from the provided command line arguments.

Parameters:

args - command line arguments

Returns:

Quadruple of configuration, execution mode and an optional listening address

createJobManagerComponents

public scala.Tuple12<ExecutorService,InstanceManager,Scheduler,BlobLibraryCacheManager,RestartStrategy,scala.concurrent.duration.FiniteDuration,Object,LeaderElectionService,SubmittedJobGraphStore,CheckpointRecoveryFactory,SavepointStore,scala.concurrent.duration.FiniteDuration> createJobManagerComponents(Configuration configuration,
                                                                                                                                                                                                                                                                                                                  scala.Option<LeaderElectionService> leaderElectionServiceOption)

Create the job manager components as (instanceManager, scheduler, libraryCacheManager, archiverProps, defaultExecutionRetries, delayBetweenRetries, timeout)

Parameters:

configuration - The configuration from which to parse the config values.

leaderElectionServiceOption - LeaderElectionService which shall be returned if the option is defined

Returns:

The members for a default JobManager.

startJobManagerActors

public scala.Tuple2<akka.actor.ActorRef,akka.actor.ActorRef> startJobManagerActors(Configuration configuration,
                                                                                   akka.actor.ActorSystem actorSystem,
                                                                                   Class<? extends JobManager> jobManagerClass,
                                                                                   Class<? extends MemoryArchivist> archiveClass)

Starts the JobManager and job archiver based on the given configuration, in the given actor system.

Parameters:

configuration - The configuration for the JobManager

actorSystem - The actor system running the JobManager

jobManagerClass - The class of the JobManager to be started

archiveClass - The class of the MemoryArchivist to be started

Returns:

A tuple of references (JobManager Ref, Archiver Ref)

startJobManagerActors

public scala.Tuple2<akka.actor.ActorRef,akka.actor.ActorRef> startJobManagerActors(Configuration configuration,
                                                                                   akka.actor.ActorSystem actorSystem,
                                                                                   scala.Option<String> jobMangerActorName,
                                                                                   scala.Option<String> archiveActorName,
                                                                                   Class<? extends JobManager> jobManagerClass,
                                                                                   Class<? extends MemoryArchivist> archiveClass)

Starts the JobManager and job archiver based on the given configuration, in the given actor system.

Parameters:

configuration - The configuration for the JobManager

actorSystem - The actor system running the JobManager

jobMangerActorName - Optionally the name of the JobManager actor. If none is given, the actor will have the name generated by the actor system.

archiveActorName - Optionally the name of the archive actor. If none is given, the actor will have the name generated by the actor system.

jobManagerClass - The class of the JobManager to be started

archiveClass - The class of the MemoryArchivist to be started

Returns:

A tuple of references (JobManager Ref, Archiver Ref)

startActor

public akka.actor.ActorRef startActor(akka.actor.Props props,
                                      akka.actor.ActorSystem actorSystem)

getRemoteJobManagerAkkaURL

public String getRemoteJobManagerAkkaURL(InetSocketAddress address,
                                         scala.Option<String> name)

Builds the akka actor path for the JobManager actor, given the socket address where the JobManager's actor system runs.

Parameters:

address - The address of the JobManager's actor system.

Returns:

The akka URL of the JobManager actor.

getRemoteJobManagerAkkaURL

public String getRemoteJobManagerAkkaURL(Configuration config)

Returns the JobManager actor's remote Akka URL, given the configured hostname and port.

Parameters:

config - The configuration to parse

Returns:

JobManager actor remote Akka URL

getLocalJobManagerAkkaURL

public String getLocalJobManagerAkkaURL(scala.Option<String> name)

Builds the akka actor path for the JobManager actor to address the actor within its own actor system.

Returns:

The local akka URL of the JobManager actor.

getJobManagerAkkaURL

public String getJobManagerAkkaURL(akka.actor.ActorSystem system,
                                   scala.Option<String> name)

getJobManagerActorRefFuture

public scala.concurrent.Future<akka.actor.ActorRef> getJobManagerActorRefFuture(InetSocketAddress address,
                                                                                akka.actor.ActorSystem system,
                                                                                scala.concurrent.duration.FiniteDuration timeout)

getJobManagerActorRef

public akka.actor.ActorRef getJobManagerActorRef(String jobManagerUrl,
                                                 akka.actor.ActorSystem system,
                                                 scala.concurrent.duration.FiniteDuration timeout)

Resolves the JobManager actor reference in a blocking fashion.

Parameters:

jobManagerUrl - The akka URL of the JobManager.

system - The local actor system that should perform the lookup.

timeout - The maximum time to wait until the lookup fails.

Returns:

The ActorRef to the JobManager

Throws:

IOException - Thrown, if the lookup fails.

getJobManagerActorRef

public akka.actor.ActorRef getJobManagerActorRef(InetSocketAddress address,
                                                 akka.actor.ActorSystem system,
                                                 scala.concurrent.duration.FiniteDuration timeout)

Resolves the JobManager actor reference in a blocking fashion.

Parameters:

address - The socket address of the JobManager's actor system.

system - The local actor system that should perform the lookup.

timeout - The maximum time to wait until the lookup fails.

Returns:

The ActorRef to the JobManager

Throws:

IOException - Thrown, if the lookup fails.

getJobManagerActorRef

public akka.actor.ActorRef getJobManagerActorRef(InetSocketAddress address,
                                                 akka.actor.ActorSystem system,
                                                 Configuration config)

Resolves the JobManager actor reference in a blocking fashion.

Parameters:

address - The socket address of the JobManager's actor system.

system - The local actor system that should perform the lookup.

config - The config describing the maximum time to wait until the lookup fails.

Returns:

The ActorRef to the JobManager

Throws:

IOException - Thrown, if the lookup fails.