org.apache.flink.client.program

Class ClusterClient

java.lang.Object

org.apache.flink.client.program.ClusterClient

Direct Known Subclasses:

RestClusterClient, StandaloneClusterClient, YarnClusterClient, YarnClusterClientV2

public abstract class ClusterClient
extends Object

Encapsulates the functionality necessary to submit a program to a remote cluster.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	ClusterClient.LazyActorSystemLoader Utility class to lazily instantiate an ActorSystem.

Field Summary

Fields

Modifier and Type	Field and Description
protected ClusterClient.LazyActorSystemLoader	actorSystemLoader The actor system used to communicate with the JobManager.
protected Configuration	flinkConfig Configuration of the client.
protected HighAvailabilityServices	highAvailabilityServices Service factory for high available.
protected org.slf4j.Logger	log
protected scala.concurrent.duration.FiniteDuration	timeout Timeout for futures.

Constructor Summary

Constructors

Constructor and Description
ClusterClient(Configuration flinkConfig) Creates a instance that submits the programs to the JobManager defined in the configuration.
ClusterClient(Configuration flinkConfig, HighAvailabilityServices highAvailabilityServices) Creates a instance that submits the programs to the JobManager defined in the configuration.

Method Summary

Modifier and Type	Method and Description
void	cancel(JobID jobId) Cancels a job identified by the job id.
String	cancelWithSavepoint(JobID jobId, String savepointDirectory) Cancels a job identified by the job id and triggers a savepoint.
JobListeningContext	connectToJob(JobID jobID) Reattaches to a running job with the given job id.
void	endSession(JobID jobId) Tells the JobManager to finish the session (job) defined by the given ID.
void	endSessions(List<JobID> jobIds) Tells the JobManager to finish the sessions (jobs) defined by the given IDs.
protected abstract void	finalizeCluster() Request the cluster to shut down or disconnect.
Map<String,Object>	getAccumulators(JobID jobID) Requests and returns the accumulators for the given job identifier.
Map<String,Object>	getAccumulators(JobID jobID, ClassLoader loader) Requests and returns the accumulators for the given job identifier.
abstract String	getClusterIdentifier() Returns a string representation of the cluster.
abstract GetClusterStatusResponse	getClusterStatus() Returns the latest cluster status, with number of Taskmanagers and slots.
Configuration	getFlinkConfiguration() Return the Flink configuration object.
JobGraph	getJobGraph(PackagedProgram prog, FlinkPlan optPlan, SavepointRestoreSettings savepointSettings)
InetSocketAddress	getJobManagerAddress() Gets the current JobManager address (may change in case of a HA setup).
ActorGateway	getJobManagerGateway() Returns the ActorGateway of the current job manager leader using the LeaderRetrievalService.
abstract int	getMaxSlots() The client may define an upper limit on the number of slots to use.
protected abstract List<String>	getNewMessages() May return new messages from the cluster.
static FlinkPlan	getOptimizedPlan(Optimizer compiler, PackagedProgram prog, int parallelism)
static OptimizedPlan	getOptimizedPlan(Optimizer compiler, Plan p, int parallelism)
static String	getOptimizedPlanAsJson(Optimizer compiler, PackagedProgram prog, int parallelism)
boolean	getPrintStatusDuringExecution()
abstract String	getWebInterfaceURL() Returns an URL (as a string) to the JobManager web interface.
abstract boolean	hasUserJarsInClassPath(List<URL> userJarFiles) Returns true if the client already has the user jar and providing it again would result in duplicate uploading of the jar.
boolean	isDetached() A flag to indicate whether this clients submits jobs detached.
CompletableFuture<Collection<JobStatusMessage>>	listJobs() Lists the currently running and finished jobs on the cluster.
protected void	logAndSysout(String message) Logs and prints to sysout if printing to stdout is enabled.
JobExecutionResult	retrieveJob(JobID jobID) Reattaches to a running from from the supplied job id.
JobSubmissionResult	run(FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths, ClassLoader classLoader)
JobSubmissionResult	run(FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths, ClassLoader classLoader, SavepointRestoreSettings savepointSettings)
JobExecutionResult	run(JobGraph jobGraph, ClassLoader classLoader) Submits a JobGraph blocking.
JobSubmissionResult	run(JobWithJars program, int parallelism)
JobSubmissionResult	run(JobWithJars jobWithJars, int parallelism, SavepointRestoreSettings savepointSettings) Runs a program on the Flink cluster to which this client is connected.
JobSubmissionResult	run(PackagedProgram prog, int parallelism) General purpose method to run a user jar from the CliFrontend in either blocking or detached mode, depending on whether setDetached(true) or setDetached(false).
JobSubmissionResult	runDetached(JobGraph jobGraph, ClassLoader classLoader) Submits a JobGraph detached.
void	setDetached(boolean isDetached) Set the mode of this client (detached or blocking job execution).
void	setPrintStatusDuringExecution(boolean print) Configures whether the client should print progress updates during the execution to System.out.
void	shutdown() Shuts down the client.
void	stop(JobID jobId) Stops a program on Flink cluster whose job-manager is configured in this client's configuration.
protected abstract JobSubmissionResult	submitJob(JobGraph jobGraph, ClassLoader classLoader) Calls the subclasses' submitJob method.
CompletableFuture<String>	triggerSavepoint(JobID jobId, String savepointDirectory) Triggers a savepoint for the job identified by the job id.
abstract void	waitForClusterToBeReady() Blocks until the client has determined that the cluster is ready for Job submission.

Methods inherited from class java.lang.Object

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

Field Detail

log

protected final org.slf4j.Logger log

actorSystemLoader

protected final ClusterClient.LazyActorSystemLoader actorSystemLoader

The actor system used to communicate with the JobManager. Lazily initialized upon first use

flinkConfig

protected final Configuration flinkConfig

Configuration of the client.

timeout

protected final scala.concurrent.duration.FiniteDuration timeout

Timeout for futures.

highAvailabilityServices

protected final HighAvailabilityServices highAvailabilityServices

Service factory for high available.

Constructor Detail

ClusterClient

public ClusterClient(Configuration flinkConfig)
              throws Exception

Creates a instance that submits the programs to the JobManager defined in the configuration. This method will try to resolve the JobManager hostname and throw an exception if that is not possible.

Parameters:

flinkConfig - The config used to obtain the job-manager's address, and used to configure the optimizer.

Throws:

Exception - we cannot create the high availability services

ClusterClient

public ClusterClient(Configuration flinkConfig,
                     HighAvailabilityServices highAvailabilityServices)

Creates a instance that submits the programs to the JobManager defined in the configuration. This method will try to resolve the JobManager hostname and throw an exception if that is not possible.

Parameters:

flinkConfig - The config used to obtain the job-manager's address, and used to configure the optimizer.

highAvailabilityServices - HighAvailabilityServices to use for leader retrieval

Method Detail

shutdown

public void shutdown()
              throws Exception

Shuts down the client. This stops the internal actor system and actors.

Throws:

Exception

setPrintStatusDuringExecution

public void setPrintStatusDuringExecution(boolean print)

Configures whether the client should print progress updates during the execution to System.out. All updates are logged via the SLF4J loggers regardless of this setting.

Parameters:

print - True to print updates to standard out during execution, false to not print them.

getPrintStatusDuringExecution

public boolean getPrintStatusDuringExecution()

Returns:

whether the client will print progress updates during the execution to System.out

getJobManagerAddress

public InetSocketAddress getJobManagerAddress()

Gets the current JobManager address (may change in case of a HA setup).

Returns:

The address (host and port) of the leading JobManager

getOptimizedPlanAsJson

public static String getOptimizedPlanAsJson(Optimizer compiler,
                                            PackagedProgram prog,
                                            int parallelism)
                                     throws CompilerException,
                                            ProgramInvocationException

Throws:

CompilerException

ProgramInvocationException

getOptimizedPlan

public static FlinkPlan getOptimizedPlan(Optimizer compiler,
                                         PackagedProgram prog,
                                         int parallelism)
                                  throws CompilerException,
                                         ProgramInvocationException

Throws:

CompilerException

ProgramInvocationException

getOptimizedPlan

public static OptimizedPlan getOptimizedPlan(Optimizer compiler,
                                             Plan p,
                                             int parallelism)
                                      throws CompilerException

Throws:

CompilerException

run

public JobSubmissionResult run(PackagedProgram prog,
                               int parallelism)
                        throws ProgramInvocationException,
                               ProgramMissingJobException

General purpose method to run a user jar from the CliFrontend in either blocking or detached mode, depending on whether setDetached(true) or setDetached(false).

Parameters:

prog - the packaged program

parallelism - the parallelism to execute the contained Flink job

Returns:

The result of the execution

Throws:

ProgramMissingJobException

ProgramInvocationException

run

public JobSubmissionResult run(JobWithJars program,
                               int parallelism)
                        throws ProgramInvocationException

Throws:

ProgramInvocationException

run

public JobSubmissionResult run(JobWithJars jobWithJars,
                               int parallelism,
                               SavepointRestoreSettings savepointSettings)
                        throws CompilerException,
                               ProgramInvocationException

Runs a program on the Flink cluster to which this client is connected. The call blocks until the execution is complete, and returns afterwards.

Parameters:

jobWithJars - The program to be executed.

parallelism - The default parallelism to use when running the program. The default parallelism is used when the program does not set a parallelism by itself.

Throws:

CompilerException - Thrown, if the compiler encounters an illegal situation.

ProgramInvocationException - Thrown, if the program could not be instantiated from its jar file, or if the submission failed. That might be either due to an I/O problem, i.e. the job-manager is unreachable, or due to the fact that the parallel execution failed.

run

public JobSubmissionResult run(FlinkPlan compiledPlan,
                               List<URL> libraries,
                               List<URL> classpaths,
                               ClassLoader classLoader)
                        throws ProgramInvocationException

Throws:

ProgramInvocationException

run

public JobSubmissionResult run(FlinkPlan compiledPlan,
                               List<URL> libraries,
                               List<URL> classpaths,
                               ClassLoader classLoader,
                               SavepointRestoreSettings savepointSettings)
                        throws ProgramInvocationException

Throws:

ProgramInvocationException

run

public JobExecutionResult run(JobGraph jobGraph,
                              ClassLoader classLoader)
                       throws ProgramInvocationException

Submits a JobGraph blocking.

Parameters:

jobGraph - The JobGraph

classLoader - User code class loader to deserialize the results and errors (may contain custom classes).

Returns:

JobExecutionResult

Throws:

ProgramInvocationException

runDetached

public JobSubmissionResult runDetached(JobGraph jobGraph,
                                       ClassLoader classLoader)
                                throws ProgramInvocationException

Submits a JobGraph detached.

Parameters:

jobGraph - The JobGraph

classLoader - User code class loader to deserialize the results and errors (may contain custom classes).

Returns:

JobSubmissionResult

Throws:

ProgramInvocationException

retrieveJob

public JobExecutionResult retrieveJob(JobID jobID)
                               throws JobExecutionException

Reattaches to a running from from the supplied job id.

Parameters:

jobID - The job id of the job to attach to

Returns:

The JobExecutionResult for the jobID

Throws:

JobExecutionException - if an error occurs during monitoring the job execution

connectToJob

public JobListeningContext connectToJob(JobID jobID)
                                 throws JobExecutionException

Reattaches to a running job with the given job id.

Parameters:

jobID - The job id of the job to attach to

Returns:

The JobExecutionResult for the jobID

Throws:

JobExecutionException - if an error occurs during monitoring the job execution

cancel

public void cancel(JobID jobId)
            throws Exception

Cancels a job identified by the job id.

Parameters:

jobId - the job id

Throws:

Exception - In case an error occurred.

cancelWithSavepoint

public String cancelWithSavepoint(JobID jobId,
                                  @Nullable
                                  String savepointDirectory)
                           throws Exception

Cancels a job identified by the job id and triggers a savepoint.

Parameters:

jobId - the job id

savepointDirectory - directory the savepoint should be written to

Returns:

path where the savepoint is located

Throws:

Exception - In case an error cocurred.

stop

public void stop(JobID jobId)
          throws Exception

Stops a program on Flink cluster whose job-manager is configured in this client's configuration. Stopping works only for streaming programs. Be aware, that the program might continue to run for a while after sending the stop command, because after sources stopped to emit data all operators need to finish processing.

Parameters:

jobId - the job ID of the streaming program to stop

Throws:

Exception - If the job ID is invalid (ie, is unknown or refers to a batch job) or if sending the stop signal failed. That might be due to an I/O problem, ie, the job-manager is unreachable.

triggerSavepoint

public CompletableFuture<String> triggerSavepoint(JobID jobId,
                                                  @Nullable
                                                  String savepointDirectory)
                                           throws Exception

Triggers a savepoint for the job identified by the job id. The savepoint will be written to the given savepoint directory, or CoreOptions.SAVEPOINT_DIRECTORY if it is null.

Parameters:

jobId - job id

savepointDirectory - directory the savepoint should be written to

Returns:

path future where the savepoint is located

Throws:

Exception - if no connection to the cluster could be established

listJobs

public CompletableFuture<Collection<JobStatusMessage>> listJobs()
                                                         throws Exception

Lists the currently running and finished jobs on the cluster.

Returns:

future collection of running and finished jobs

Throws:

Exception - if no connection to the cluster could be established

getAccumulators

public Map<String,Object> getAccumulators(JobID jobID)
                                   throws Exception

Requests and returns the accumulators for the given job identifier. Accumulators can be requested while a is running or after it has finished. The default class loader is used to deserialize the incoming accumulator results.

Parameters:

jobID - The job identifier of a job.

Returns:

A Map containing the accumulator's name and its value.

Throws:

Exception

getAccumulators

public Map<String,Object> getAccumulators(JobID jobID,
                                          ClassLoader loader)
                                   throws Exception

Requests and returns the accumulators for the given job identifier. Accumulators can be requested while a is running or after it has finished.

Parameters:

jobID - The job identifier of a job.

loader - The class loader for deserializing the accumulator results.

Returns:

A Map containing the accumulator's name and its value.

Throws:

Exception

endSession

public void endSession(JobID jobId)
                throws Exception

Tells the JobManager to finish the session (job) defined by the given ID.

Parameters:

jobId - The ID that identifies the session.

Throws:

Exception

endSessions

public void endSessions(List<JobID> jobIds)
                 throws Exception

Tells the JobManager to finish the sessions (jobs) defined by the given IDs.

Parameters:

jobIds - The IDs that identify the sessions.

Throws:

Exception

getJobGraph

public JobGraph getJobGraph(PackagedProgram prog,
                            FlinkPlan optPlan,
                            SavepointRestoreSettings savepointSettings)
                     throws ProgramInvocationException

Throws:

ProgramInvocationException

getJobManagerGateway

public ActorGateway getJobManagerGateway()
                                  throws Exception

Returns the ActorGateway of the current job manager leader using the LeaderRetrievalService.

Returns:

ActorGateway of the current job manager leader

Throws:

Exception

logAndSysout

protected void logAndSysout(String message)

Logs and prints to sysout if printing to stdout is enabled.

Parameters:

message - The message to log/print

waitForClusterToBeReady

public abstract void waitForClusterToBeReady()

Blocks until the client has determined that the cluster is ready for Job submission.

This is delayed until right before job submission to report any other errors first (e.g. invalid job definitions/errors in the user jar)

getWebInterfaceURL

public abstract String getWebInterfaceURL()

Returns an URL (as a string) to the JobManager web interface.

getClusterStatus

public abstract GetClusterStatusResponse getClusterStatus()

Returns the latest cluster status, with number of Taskmanagers and slots.

getNewMessages

protected abstract List<String> getNewMessages()

May return new messages from the cluster. Messages can be for example about failed containers or container launch requests.

getClusterIdentifier

public abstract String getClusterIdentifier()

Returns a string representation of the cluster.

finalizeCluster

protected abstract void finalizeCluster()

Request the cluster to shut down or disconnect.

setDetached

public void setDetached(boolean isDetached)

Set the mode of this client (detached or blocking job execution).

Parameters:

isDetached - If true, the client will submit programs detached via the run method

isDetached

public boolean isDetached()

A flag to indicate whether this clients submits jobs detached.

Returns:

True if the Client submits detached, false otherwise

getFlinkConfiguration

public Configuration getFlinkConfiguration()

Return the Flink configuration object.

Returns:

The Flink configuration object

getMaxSlots

public abstract int getMaxSlots()

The client may define an upper limit on the number of slots to use.

Returns:

-1 if unknown

hasUserJarsInClassPath

public abstract boolean hasUserJarsInClassPath(List<URL> userJarFiles)

Returns true if the client already has the user jar and providing it again would result in duplicate uploading of the jar.

submitJob

protected abstract JobSubmissionResult submitJob(JobGraph jobGraph,
                                                 ClassLoader classLoader)
                                          throws ProgramInvocationException

Calls the subclasses' submitJob method. It may decide to simply call one of the run methods or it may perform some custom job submission logic.

Parameters:

jobGraph - The JobGraph to be submitted

Returns:

JobSubmissionResult

Throws:

ProgramInvocationException