org.apache.flink.runtime.minicluster

Class MiniCluster

java.lang.Object

org.apache.flink.runtime.minicluster.MiniCluster

All Implemented Interfaces:

AutoCloseable, AutoCloseableAsync

public class MiniCluster
extends Object
implements AutoCloseableAsync

MiniCluster to execute Flink jobs locally.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	MiniCluster.CommonRpcServiceFactory Factory which returns always the common RpcService.
protected class	MiniCluster.DedicatedRpcServiceFactory Factory which creates and registers new RpcService.
static class	MiniCluster.HaServices HA Services to use.
protected static interface	MiniCluster.RpcServiceFactory Internal factory for RpcService.

Constructor Summary

Constructors

Constructor and Description
MiniCluster(MiniClusterConfiguration miniClusterConfiguration) Creates a new Flink mini cluster based on the given configuration.
MiniCluster(MiniClusterConfiguration miniClusterConfiguration, Supplier<Reference<RpcSystem>> rpcSystemSupplier)

Method Summary

Modifier and Type	Method and Description
CompletableFuture<Acknowledge>	cancelJob(JobID jobId)
CompletableFuture<Void>	closeAsync() Shuts down the mini cluster, failing all currently executing jobs.
CompletableFuture<Void>	closeAsyncWithoutCleaningHighAvailabilityData()
protected DispatcherResourceManagerComponentFactory	createDispatcherResourceManagerComponentFactory()
protected Collection<? extends DispatcherResourceManagerComponent>	createDispatcherResourceManagerComponents(Configuration configuration, MiniCluster.RpcServiceFactory rpcServiceFactory, BlobServer blobServer, HeartbeatServices heartbeatServices, DelegationTokenManager delegationTokenManager, MetricRegistry metricRegistry, MetricQueryServiceRetriever metricQueryServiceRetriever, FatalErrorHandler fatalErrorHandler)
protected HighAvailabilityServices	createHighAvailabilityServices(Configuration configuration, Executor executor)
protected RpcService	createLocalRpcService(Configuration configuration, RpcSystem rpcSystem) Factory method to instantiate the local RPC service.
protected MetricRegistryImpl	createMetricRegistry(Configuration config, long maximumMessageSizeInBytes) Factory method to create the metric registry for the mini cluster.
protected RpcService	createRemoteRpcService(Configuration configuration, String bindAddress, int bindPort, RpcSystem rpcSystem) Factory method to instantiate the remote RPC service.
protected RpcService	createRemoteRpcService(Configuration configuration, String externalAddress, String externalPortRange, String bindAddress, RpcSystem rpcSystem) Factory method to instantiate the remote RPC service.
CompletableFuture<CoordinationResponse>	deliverCoordinationRequestToCoordinator(JobID jobId, OperatorID operatorId, SerializedValue<CoordinationRequest> serializedRequest)
CompletableFuture<Acknowledge>	disposeSavepoint(String savepointPath)
JobExecutionResult	executeJobBlocking(JobGraph job) This method runs a job in blocking mode.
CompletableFuture<ArchivedExecutionGraph>	getArchivedExecutionGraph(JobID jobId)
ClusterInformation	getClusterInformation()
Configuration	getConfiguration()
protected CompletableFuture<DispatcherGateway>	getDispatcherGatewayFuture()
CompletableFuture<? extends AccessExecutionGraph>	getExecutionGraph(JobID jobId)
Optional<HaLeadershipControl>	getHaLeadershipControl() Returns HaLeadershipControl if enabled.
protected HighAvailabilityServices	getHaServices()
protected Executor	getIOExecutor()
CompletableFuture<JobStatus>	getJobStatus(JobID jobId)
CompletableFuture<ResourceOverview>	getResourceOverview()
CompletableFuture<URI>	getRestAddress()
CompletableFuture<Void>	invalidateClusterDataset(AbstractID clusterDatasetId)
boolean	isRunning() Checks if the mini cluster was started and is running.
CompletableFuture<Set<AbstractID>>	listCompletedClusterDatasetIds()
CompletableFuture<Collection<JobStatusMessage>>	listJobs()
void	overrideRestoreModeForChangelogStateBackend()
CompletableFuture<Void>	reportHeartbeat(JobID jobId, long expiredTimestamp)
CompletableFuture<ClusterOverview>	requestClusterOverview()
CompletableFuture<JobResult>	requestJobResult(JobID jobId)
void	runDetached(JobGraph job) This method executes a job in detached mode.
void	start() Starts the mini cluster, based on the configured properties.
void	startTaskManager() Starts additional TaskManager process.
CompletableFuture<String>	stopWithDetachedSavepoint(JobID jobId, String targetDirectory, boolean terminate, SavepointFormatType formatType)
CompletableFuture<String>	stopWithSavepoint(JobID jobId, String targetDirectory, boolean terminate, SavepointFormatType formatType)
CompletableFuture<JobSubmissionResult>	submitJob(JobGraph jobGraph)
CompletableFuture<Void>	terminateTaskManager(int index) Terminates a TaskManager with the given index.
CompletableFuture<String>	triggerCheckpoint(JobID jobID)
CompletableFuture<Long>	triggerCheckpoint(JobID jobID, CheckpointType checkpointType)
CompletableFuture<String>	triggerDetachedSavepoint(JobID jobId, String targetDirectory, boolean cancelJob, SavepointFormatType formatType)
CompletableFuture<String>	triggerSavepoint(JobID jobId, String targetDirectory, boolean cancelJob, SavepointFormatType formatType)
protected boolean	useLocalCommunication()

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.util.AutoCloseableAsync

close

Constructor Detail

MiniCluster

public MiniCluster(MiniClusterConfiguration miniClusterConfiguration)

Creates a new Flink mini cluster based on the given configuration.

Parameters:

miniClusterConfiguration - The configuration for the mini cluster

MiniCluster

public MiniCluster(MiniClusterConfiguration miniClusterConfiguration,
                   Supplier<Reference<RpcSystem>> rpcSystemSupplier)

Method Detail

getRestAddress

public CompletableFuture<URI> getRestAddress()

getClusterInformation

public ClusterInformation getClusterInformation()

getIOExecutor

protected Executor getIOExecutor()

isRunning

public boolean isRunning()

Checks if the mini cluster was started and is running.

start

public void start()
           throws Exception

Starts the mini cluster, based on the configured properties.

Throws:

Exception - This method passes on any exception that occurs during the startup of the mini cluster.

createDispatcherResourceManagerComponents

@VisibleForTesting
protected Collection<? extends DispatcherResourceManagerComponent> createDispatcherResourceManagerComponents(Configuration configuration,
                                                                                                                                MiniCluster.RpcServiceFactory rpcServiceFactory,
                                                                                                                                BlobServer blobServer,
                                                                                                                                HeartbeatServices heartbeatServices,
                                                                                                                                DelegationTokenManager delegationTokenManager,
                                                                                                                                MetricRegistry metricRegistry,
                                                                                                                                MetricQueryServiceRetriever metricQueryServiceRetriever,
                                                                                                                                FatalErrorHandler fatalErrorHandler)
                                                                                                                         throws Exception

Throws:

Exception

createDispatcherResourceManagerComponentFactory

protected DispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory()

createHighAvailabilityServices

@VisibleForTesting
protected HighAvailabilityServices createHighAvailabilityServices(Configuration configuration,
                                                                                     Executor executor)
                                                                              throws Exception

Throws:

Exception

getHaLeadershipControl

public Optional<HaLeadershipControl> getHaLeadershipControl()

Returns HaLeadershipControl if enabled.

HaLeadershipControl allows granting and revoking leadership of HA components, e.g. JobManager. The method return Optional.empty() if the control is not enabled in MiniClusterConfiguration.

Enabling this feature disables HighAvailabilityOptions.HA_MODE option.

getHaServices

protected HighAvailabilityServices getHaServices()

closeAsync

public CompletableFuture<Void> closeAsync()

Shuts down the mini cluster, failing all currently executing jobs. The mini cluster can be started again by calling the start() method again.

This method shuts down all started services and components, even if an exception occurs in the process of shutting down some component.

Specified by:

closeAsync in interface AutoCloseableAsync

Returns:

Future which is completed once the MiniCluster has been completely shut down

closeAsyncWithoutCleaningHighAvailabilityData

public CompletableFuture<Void> closeAsyncWithoutCleaningHighAvailabilityData()

startTaskManager

public void startTaskManager()
                      throws Exception

Starts additional TaskManager process.

When the MiniCluster starts up, it always starts MiniClusterConfiguration.getNumTaskManagers() TaskManagers. All TaskManagers are indexed from 0 to the number of TaskManagers, started so far, minus one. This method starts a TaskManager with the next index which is the number of TaskManagers, started so far. The index always increases with each new started TaskManager. The indices of terminated TaskManagers are not reused after terminateTaskManager(int).

Throws:

Exception

useLocalCommunication

@VisibleForTesting
protected boolean useLocalCommunication()

getConfiguration

@VisibleForTesting
public Configuration getConfiguration()

overrideRestoreModeForChangelogStateBackend

@Internal
public void overrideRestoreModeForChangelogStateBackend()

terminateTaskManager

public CompletableFuture<Void> terminateTaskManager(int index)

Terminates a TaskManager with the given index.

See startTaskManager() to understand how TaskManagers are indexed. This method terminates a TaskManager with a given index but it does not clear the index. The index stays occupied for the lifetime of the MiniCluster and its TaskManager stays terminated. The index is not reused if more TaskManagers are started with startTaskManager().

Parameters:

index - index of the TaskManager to terminate

Returns:

CompletableFuture of the given TaskManager termination

getArchivedExecutionGraph

public CompletableFuture<ArchivedExecutionGraph> getArchivedExecutionGraph(JobID jobId)

listJobs

public CompletableFuture<Collection<JobStatusMessage>> listJobs()

getJobStatus

public CompletableFuture<JobStatus> getJobStatus(JobID jobId)

cancelJob

public CompletableFuture<Acknowledge> cancelJob(JobID jobId)

triggerSavepoint

public CompletableFuture<String> triggerSavepoint(JobID jobId,
                                                  String targetDirectory,
                                                  boolean cancelJob,
                                                  SavepointFormatType formatType)

triggerDetachedSavepoint

public CompletableFuture<String> triggerDetachedSavepoint(JobID jobId,
                                                          String targetDirectory,
                                                          boolean cancelJob,
                                                          SavepointFormatType formatType)

triggerCheckpoint

public CompletableFuture<String> triggerCheckpoint(JobID jobID)

triggerCheckpoint

public CompletableFuture<Long> triggerCheckpoint(JobID jobID,
                                                 CheckpointType checkpointType)

stopWithSavepoint

public CompletableFuture<String> stopWithSavepoint(JobID jobId,
                                                   String targetDirectory,
                                                   boolean terminate,
                                                   SavepointFormatType formatType)

stopWithDetachedSavepoint

public CompletableFuture<String> stopWithDetachedSavepoint(JobID jobId,
                                                           String targetDirectory,
                                                           boolean terminate,
                                                           SavepointFormatType formatType)

disposeSavepoint

public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath)

getExecutionGraph

public CompletableFuture<? extends AccessExecutionGraph> getExecutionGraph(JobID jobId)

deliverCoordinationRequestToCoordinator

public CompletableFuture<CoordinationResponse> deliverCoordinationRequestToCoordinator(JobID jobId,
                                                                                       OperatorID operatorId,
                                                                                       SerializedValue<CoordinationRequest> serializedRequest)

getResourceOverview

public CompletableFuture<ResourceOverview> getResourceOverview()

runDetached

public void runDetached(JobGraph job)
                 throws JobExecutionException,
                        InterruptedException

This method executes a job in detached mode. The method returns immediately after the job has been added to the

Parameters:

job - The Flink job to execute

Throws:

JobExecutionException - Thrown if anything went amiss during initial job launch, or if the job terminally failed.

InterruptedException

executeJobBlocking

public JobExecutionResult executeJobBlocking(JobGraph job)
                                      throws JobExecutionException,
                                             InterruptedException

This method runs a job in blocking mode. The method returns only after the job completed successfully, or after it failed terminally.

Parameters:

job - The Flink job to execute

Returns:

The result of the job execution

Throws:

JobExecutionException - Thrown if anything went amiss during initial job launch, or if the job terminally failed.

InterruptedException

submitJob

public CompletableFuture<JobSubmissionResult> submitJob(JobGraph jobGraph)

requestJobResult

public CompletableFuture<JobResult> requestJobResult(JobID jobId)

requestClusterOverview

public CompletableFuture<ClusterOverview> requestClusterOverview()

getDispatcherGatewayFuture

@VisibleForTesting
protected CompletableFuture<DispatcherGateway> getDispatcherGatewayFuture()

createMetricRegistry

protected MetricRegistryImpl createMetricRegistry(Configuration config,
                                                  long maximumMessageSizeInBytes)

Factory method to create the metric registry for the mini cluster.

Parameters:

config - The configuration of the mini cluster

maximumMessageSizeInBytes - the maximum message size

createRemoteRpcService

protected RpcService createRemoteRpcService(Configuration configuration,
                                            String bindAddress,
                                            int bindPort,
                                            RpcSystem rpcSystem)
                                     throws Exception

Factory method to instantiate the remote RPC service.

Parameters:

configuration - Flink configuration.

bindAddress - The address to bind the RPC service to.

bindPort - The port range to bind the RPC service to.

rpcSystem -

Returns:

The instantiated RPC service

Throws:

Exception

createRemoteRpcService

protected RpcService createRemoteRpcService(Configuration configuration,
                                            String externalAddress,
                                            String externalPortRange,
                                            String bindAddress,
                                            RpcSystem rpcSystem)
                                     throws Exception

Factory method to instantiate the remote RPC service.

Parameters:

configuration - Flink configuration.

externalAddress - The external address to access the RPC service.

externalPortRange - The external port range to access the RPC service.

bindAddress - The address to bind the RPC service to.

rpcSystem -

Returns:

The instantiated RPC service

Throws:

Exception

createLocalRpcService

protected RpcService createLocalRpcService(Configuration configuration,
                                           RpcSystem rpcSystem)
                                    throws Exception

Factory method to instantiate the local RPC service.

Parameters:

configuration - Flink configuration.

rpcSystem -

Returns:

The instantiated RPC service

Throws:

Exception

invalidateClusterDataset

public CompletableFuture<Void> invalidateClusterDataset(AbstractID clusterDatasetId)

listCompletedClusterDatasetIds

public CompletableFuture<Set<AbstractID>> listCompletedClusterDatasetIds()

reportHeartbeat

public CompletableFuture<Void> reportHeartbeat(JobID jobId,
                                               long expiredTimestamp)