org.apache.flink.runtime.scheduler.adaptive

Class AdaptiveScheduler

java.lang.Object

org.apache.flink.runtime.scheduler.adaptive.AdaptiveScheduler

All Implemented Interfaces:

AutoCloseable, GlobalFailureHandler, SchedulerNG, AutoCloseableAsync

public class AdaptiveScheduler
extends Object
implements SchedulerNG

A SchedulerNG implementation that uses the declarative resource management and automatically adapts the parallelism in case not enough resource could be acquired to run at the configured parallelism, as described in FLIP-160.

This scheduler only supports jobs with streaming semantics, i.e., all vertices are connected via pipelined data-exchanges.

The implementation is spread over multiple State classes that control which RPCs are allowed in a given state and what state transitions are possible (see the FLIP for an overview). This class can thus be roughly split into 2 parts:

1) RPCs, which must forward the call to the state via State.tryRun(Class, ThrowingConsumer, String) or State.tryCall(Class, FunctionWithException, String).

2) Context methods, which are called by states, to either transition into another state or access functionality of some component in the scheduler.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	AdaptiveScheduler.Settings Consolidated settings for the adaptive scheduler.

Constructor Summary

Constructors

Constructor and Description

AdaptiveScheduler(AdaptiveScheduler.Settings settings, JobGraph jobGraph, JobResourceRequirements jobResourceRequirements, Configuration configuration, DeclarativeSlotPool declarativeSlotPool, SlotAllocator slotAllocator, Executor ioExecutor, ClassLoader userCodeClassLoader, CheckpointsCleaner checkpointsCleaner, CheckpointRecoveryFactory checkpointRecoveryFactory, JobManagerJobMetricGroup jobManagerJobMetricGroup, RestartBackoffTimeStrategy restartBackoffTimeStrategy, long initializationTimestamp, ComponentMainThreadExecutor mainThreadExecutor, FatalErrorHandler fatalErrorHandler, JobStatusListener jobStatusListener, Collection<FailureEnricher> failureEnrichers, ExecutionGraphFactory executionGraphFactory)

Method Summary

Modifier and Type	Method and Description
void	acknowledgeCheckpoint(JobID jobID, ExecutionAttemptID executionAttemptID, long checkpointId, CheckpointMetrics checkpointMetrics, TaskStateSnapshot checkpointState)
void	archiveFailure(RootExceptionHistoryEntry failure) Archive failure.
void	cancel()
CompletableFuture<Void>	closeAsync() Trigger the closing of the resource and return the corresponding close future.
void	declineCheckpoint(DeclineCheckpoint decline)
CompletableFuture<CoordinationResponse>	deliverCoordinationRequestToCoordinator(OperatorID operator, CoordinationRequest request) Delivers a coordination request to the OperatorCoordinator with the given OperatorID and returns the coordinator's response.
void	deliverOperatorEventToCoordinator(ExecutionAttemptID taskExecution, OperatorID operator, OperatorEvent evt) Delivers the given OperatorEvent to the OperatorCoordinator with the given OperatorID.
ArchivedExecutionGraph	getArchivedExecutionGraph(JobStatus jobStatus, Throwable cause) Creates the ArchivedExecutionGraph for the given job status and cause.
Executor	getIOExecutor() Gets the I/O executor.
CompletableFuture<JobStatus>	getJobTerminationFuture()
ComponentMainThreadExecutor	getMainThreadExecutor() Gets the main thread executor.
JobManagerJobMetricGroup	getMetricGroup() Gets the JobManagerJobMetricGroup.
void	goToCanceling(ExecutionGraph executionGraph, ExecutionGraphHandler executionGraphHandler, OperatorCoordinatorHandler operatorCoordinatorHandler, List<ExceptionHistoryEntry> failureCollection) Transitions into the Canceling state.
void	goToCreatingExecutionGraph(ExecutionGraph previousExecutionGraph) Transitions into the CreatingExecutionGraph state.
void	goToExecuting(ExecutionGraph executionGraph, ExecutionGraphHandler executionGraphHandler, OperatorCoordinatorHandler operatorCoordinatorHandler, List<ExceptionHistoryEntry> failureCollection) Transitions into the Executing state.
void	goToFailing(ExecutionGraph executionGraph, ExecutionGraphHandler executionGraphHandler, OperatorCoordinatorHandler operatorCoordinatorHandler, Throwable failureCause, List<ExceptionHistoryEntry> failureCollection) Transitions into the Failing state.
void	goToFinished(ArchivedExecutionGraph archivedExecutionGraph) Transitions into the Finished state.
void	goToRestarting(ExecutionGraph executionGraph, ExecutionGraphHandler executionGraphHandler, OperatorCoordinatorHandler operatorCoordinatorHandler, Duration backoffTime, List<ExceptionHistoryEntry> failureCollection) Transitions into the Restarting state.
CompletableFuture<String>	goToStopWithSavepoint(ExecutionGraph executionGraph, ExecutionGraphHandler executionGraphHandler, OperatorCoordinatorHandler operatorCoordinatorHandler, CheckpointScheduling checkpointScheduling, CompletableFuture<String> savepointFuture, List<ExceptionHistoryEntry> failureCollection) Transitions into the StopWithSavepoint state.
void	goToWaitingForResources(ExecutionGraph previousExecutionGraph) Transitions into the WaitingForResources state.
void	handleGlobalFailure(Throwable cause) Handles a global failure.
boolean	hasDesiredResources() Checks whether we have the desired resources.
boolean	hasSufficientResources() Checks if we currently have sufficient resources for executing the job.
org.apache.flink.runtime.scheduler.adaptive.FailureResult	howToHandleFailure(Throwable failure, CompletableFuture<Map<String,String>> failureLabels) Asks how to handle the failure.
boolean	isState(org.apache.flink.runtime.scheduler.adaptive.State expectedState) Checks whether the current state is the expected state.
void	notifyEndOfData(ExecutionAttemptID executionAttemptID) Notifies that the task has reached the end of data.
void	notifyKvStateRegistered(JobID jobId, JobVertexID jobVertexId, KeyGroupRange keyGroupRange, String registrationName, KvStateID kvStateId, InetSocketAddress kvStateServerAddress)
void	notifyKvStateUnregistered(JobID jobId, JobVertexID jobVertexId, KeyGroupRange keyGroupRange, String registrationName)
void	onFinished(ArchivedExecutionGraph archivedExecutionGraph) Callback which is called when the execution reaches the Finished state.
void	reportCheckpointMetrics(JobID jobID, ExecutionAttemptID executionAttemptID, long checkpointId, CheckpointMetrics checkpointMetrics)
void	reportInitializationMetrics(JobID jobId, SubTaskInitializationMetrics initializationMetrics)
CheckpointStatsSnapshot	requestCheckpointStats() Returns the checkpoint statistics for a given job.
ExecutionGraphInfo	requestJob()
JobResourceRequirements	requestJobResourceRequirements() Read current job resource requirements.
JobStatus	requestJobStatus()
KvStateLocation	requestKvStateLocation(JobID jobId, String registrationName)
SerializedInputSplit	requestNextInputSplit(JobVertexID vertexID, ExecutionAttemptID executionAttempt)
ExecutionState	requestPartitionState(IntermediateDataSetID intermediateResultId, ResultPartitionID resultPartitionId)
void	runIfState(org.apache.flink.runtime.scheduler.adaptive.State expectedState, Runnable action) Run the given action if the current state equals the expected state.
ScheduledFuture<?>	runIfState(org.apache.flink.runtime.scheduler.adaptive.State expectedState, Runnable action, Duration delay) Runs the given action after a delay if the state at this time equals the expected state.
boolean	shouldRescale(ExecutionGraph executionGraph, boolean forceRescale) In regular mode, rescale the job if added resource meets JobManagerOptions.MIN_PARALLELISM_INCREASE.
void	startScheduling()
CompletableFuture<String>	stopWithSavepoint(String targetDirectory, boolean terminate, SavepointFormatType formatType)
CompletableFuture<CompletedCheckpoint>	triggerCheckpoint(CheckpointType checkpointType)
CompletableFuture<String>	triggerSavepoint(String targetDirectory, boolean cancelJob, SavepointFormatType formatType)
org.apache.flink.runtime.scheduler.adaptive.CreatingExecutionGraph.AssignmentResult	tryToAssignSlots(org.apache.flink.runtime.scheduler.adaptive.CreatingExecutionGraph.ExecutionGraphWithVertexParallelism executionGraphWithVertexParallelism) Try to assign slots to the created ExecutionGraph.
void	updateAccumulators(AccumulatorSnapshot accumulatorSnapshot)
void	updateJobResourceRequirements(JobResourceRequirements jobResourceRequirements) Update job resource requirements.
boolean	updateTaskExecutionState(TaskExecutionStateTransition taskExecutionState)

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.scheduler.SchedulerNG

updateTaskExecutionState

Methods inherited from interface org.apache.flink.util.AutoCloseableAsync

close

Constructor Detail

AdaptiveScheduler

public AdaptiveScheduler(AdaptiveScheduler.Settings settings,
                         JobGraph jobGraph,
                         @Nullable
                         JobResourceRequirements jobResourceRequirements,
                         Configuration configuration,
                         DeclarativeSlotPool declarativeSlotPool,
                         SlotAllocator slotAllocator,
                         Executor ioExecutor,
                         ClassLoader userCodeClassLoader,
                         CheckpointsCleaner checkpointsCleaner,
                         CheckpointRecoveryFactory checkpointRecoveryFactory,
                         JobManagerJobMetricGroup jobManagerJobMetricGroup,
                         RestartBackoffTimeStrategy restartBackoffTimeStrategy,
                         long initializationTimestamp,
                         ComponentMainThreadExecutor mainThreadExecutor,
                         FatalErrorHandler fatalErrorHandler,
                         JobStatusListener jobStatusListener,
                         Collection<FailureEnricher> failureEnrichers,
                         ExecutionGraphFactory executionGraphFactory)
                  throws JobExecutionException

Throws:

JobExecutionException

Method Detail

startScheduling

public void startScheduling()

Specified by:

startScheduling in interface SchedulerNG

closeAsync

public CompletableFuture<Void> closeAsync()

Description copied from interface: AutoCloseableAsync

Trigger the closing of the resource and return the corresponding close future.

Specified by:

closeAsync in interface AutoCloseableAsync

Returns:

Future which is completed once the resource has been closed

cancel

public void cancel()

Specified by:

cancel in interface SchedulerNG

getJobTerminationFuture

public CompletableFuture<JobStatus> getJobTerminationFuture()

Specified by:

getJobTerminationFuture in interface SchedulerNG

handleGlobalFailure

public void handleGlobalFailure(Throwable cause)

Description copied from interface: GlobalFailureHandler

Handles a global failure.

Specified by:

handleGlobalFailure in interface GlobalFailureHandler

Parameters:

cause - A cause that describes the global failure.

updateTaskExecutionState

public boolean updateTaskExecutionState(TaskExecutionStateTransition taskExecutionState)

Specified by:

updateTaskExecutionState in interface SchedulerNG

requestNextInputSplit

public SerializedInputSplit requestNextInputSplit(JobVertexID vertexID,
                                                  ExecutionAttemptID executionAttempt)
                                           throws IOException

Specified by:

requestNextInputSplit in interface SchedulerNG

Throws:

IOException

requestPartitionState

public ExecutionState requestPartitionState(IntermediateDataSetID intermediateResultId,
                                            ResultPartitionID resultPartitionId)
                                     throws PartitionProducerDisposedException

Specified by:

requestPartitionState in interface SchedulerNG

Throws:

PartitionProducerDisposedException

requestJob

public ExecutionGraphInfo requestJob()

Specified by:

requestJob in interface SchedulerNG

requestCheckpointStats

public CheckpointStatsSnapshot requestCheckpointStats()

Description copied from interface: SchedulerNG

Returns the checkpoint statistics for a given job. Although the CheckpointStatsSnapshot is included in the ExecutionGraphInfo, this method is preferred to SchedulerNG.requestJob() because it is less expensive.

Specified by:

requestCheckpointStats in interface SchedulerNG

Returns:

checkpoint statistics snapshot for job graph

archiveFailure

public void archiveFailure(RootExceptionHistoryEntry failure)

Archive failure.

requestJobStatus

public JobStatus requestJobStatus()

Specified by:

requestJobStatus in interface SchedulerNG

requestKvStateLocation

public KvStateLocation requestKvStateLocation(JobID jobId,
                                              String registrationName)
                                       throws UnknownKvStateLocation,
                                              FlinkJobNotFoundException

Specified by:

requestKvStateLocation in interface SchedulerNG

Throws:

UnknownKvStateLocation

FlinkJobNotFoundException

notifyKvStateRegistered

public void notifyKvStateRegistered(JobID jobId,
                                    JobVertexID jobVertexId,
                                    KeyGroupRange keyGroupRange,
                                    String registrationName,
                                    KvStateID kvStateId,
                                    InetSocketAddress kvStateServerAddress)
                             throws FlinkJobNotFoundException

Specified by:

notifyKvStateRegistered in interface SchedulerNG

Throws:

FlinkJobNotFoundException

notifyKvStateUnregistered

public void notifyKvStateUnregistered(JobID jobId,
                                      JobVertexID jobVertexId,
                                      KeyGroupRange keyGroupRange,
                                      String registrationName)
                               throws FlinkJobNotFoundException

Specified by:

notifyKvStateUnregistered in interface SchedulerNG

Throws:

FlinkJobNotFoundException

updateAccumulators

public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot)

Specified by:

updateAccumulators in interface SchedulerNG

triggerSavepoint

public CompletableFuture<String> triggerSavepoint(@Nullable
                                                  String targetDirectory,
                                                  boolean cancelJob,
                                                  SavepointFormatType formatType)

Specified by:

triggerSavepoint in interface SchedulerNG

triggerCheckpoint

public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(CheckpointType checkpointType)

Specified by:

triggerCheckpoint in interface SchedulerNG

acknowledgeCheckpoint

public void acknowledgeCheckpoint(JobID jobID,
                                  ExecutionAttemptID executionAttemptID,
                                  long checkpointId,
                                  CheckpointMetrics checkpointMetrics,
                                  TaskStateSnapshot checkpointState)

Specified by:

acknowledgeCheckpoint in interface SchedulerNG

notifyEndOfData

public void notifyEndOfData(ExecutionAttemptID executionAttemptID)

Description copied from interface: SchedulerNG

Notifies that the task has reached the end of data.

Specified by:

notifyEndOfData in interface SchedulerNG

Parameters:

executionAttemptID - The execution attempt id.

reportCheckpointMetrics

public void reportCheckpointMetrics(JobID jobID,
                                    ExecutionAttemptID executionAttemptID,
                                    long checkpointId,
                                    CheckpointMetrics checkpointMetrics)

Specified by:

reportCheckpointMetrics in interface SchedulerNG

declineCheckpoint

public void declineCheckpoint(DeclineCheckpoint decline)

Specified by:

declineCheckpoint in interface SchedulerNG

reportInitializationMetrics

public void reportInitializationMetrics(JobID jobId,
                                        SubTaskInitializationMetrics initializationMetrics)

Specified by:

reportInitializationMetrics in interface SchedulerNG

stopWithSavepoint

public CompletableFuture<String> stopWithSavepoint(@Nullable
                                                   String targetDirectory,
                                                   boolean terminate,
                                                   SavepointFormatType formatType)

Specified by:

stopWithSavepoint in interface SchedulerNG

deliverOperatorEventToCoordinator

public void deliverOperatorEventToCoordinator(ExecutionAttemptID taskExecution,
                                              OperatorID operator,
                                              OperatorEvent evt)
                                       throws FlinkException

Description copied from interface: SchedulerNG

Delivers the given OperatorEvent to the OperatorCoordinator with the given OperatorID.

Failure semantics: If the task manager sends an event for a non-running task or a non-existing operator coordinator, then respond with an exception to the call. If task and coordinator exist, then we assume that the call from the TaskManager was valid, and any bubbling exception needs to cause a job failure

Specified by:

deliverOperatorEventToCoordinator in interface SchedulerNG

Throws:

FlinkException - Thrown, if the task is not running or no operator/coordinator exists for the given ID.

deliverCoordinationRequestToCoordinator

public CompletableFuture<CoordinationResponse> deliverCoordinationRequestToCoordinator(OperatorID operator,
                                                                                       CoordinationRequest request)
                                                                                throws FlinkException

Description copied from interface: SchedulerNG

Delivers a coordination request to the OperatorCoordinator with the given OperatorID and returns the coordinator's response.

Specified by:

deliverCoordinationRequestToCoordinator in interface SchedulerNG

Returns:

A future containing the response.

Throws:

FlinkException - Thrown, if the task is not running, or no operator/coordinator exists for the given ID, or the coordinator cannot handle client events.

requestJobResourceRequirements

public JobResourceRequirements requestJobResourceRequirements()

Description copied from interface: SchedulerNG

Read current job resource requirements.

Specified by:

requestJobResourceRequirements in interface SchedulerNG

Returns:

Current resource requirements.

updateJobResourceRequirements

public void updateJobResourceRequirements(JobResourceRequirements jobResourceRequirements)

Description copied from interface: SchedulerNG

Update job resource requirements.

Specified by:

updateJobResourceRequirements in interface SchedulerNG

Parameters:

jobResourceRequirements - new resource requirements

hasDesiredResources

public boolean hasDesiredResources()

Checks whether we have the desired resources.

Returns:

true if we have enough resources; otherwise false

hasSufficientResources

public boolean hasSufficientResources()

Checks if we currently have sufficient resources for executing the job.

Returns:

true if we have sufficient resources; otherwise false

getArchivedExecutionGraph

public ArchivedExecutionGraph getArchivedExecutionGraph(JobStatus jobStatus,
                                                        @Nullable
                                                        Throwable cause)

Creates the ArchivedExecutionGraph for the given job status and cause. Cause can be null if there is no failure.

Parameters:

jobStatus - jobStatus to initialize the ArchivedExecutionGraph with

cause - cause describing a failure cause; null if there is none

Returns:

the created ArchivedExecutionGraph

goToWaitingForResources

public void goToWaitingForResources(@Nullable
                                    ExecutionGraph previousExecutionGraph)

Description copied from interface: StateTransitions.ToWaitingForResources

Transitions into the WaitingForResources state.

goToExecuting

public void goToExecuting(ExecutionGraph executionGraph,
                          ExecutionGraphHandler executionGraphHandler,
                          OperatorCoordinatorHandler operatorCoordinatorHandler,
                          List<ExceptionHistoryEntry> failureCollection)

Description copied from interface: StateTransitions.ToExecuting

Transitions into the Executing state.

Parameters:

executionGraph - executionGraph to pass to the Executing state

executionGraphHandler - executionGraphHandler to pass to the Executing state

operatorCoordinatorHandler - operatorCoordinatorHandler to pass to the Executing state

failureCollection - collection of failures that are propagated

goToCanceling

public void goToCanceling(ExecutionGraph executionGraph,
                          ExecutionGraphHandler executionGraphHandler,
                          OperatorCoordinatorHandler operatorCoordinatorHandler,
                          List<ExceptionHistoryEntry> failureCollection)

Description copied from interface: StateTransitions.ToCancelling

Transitions into the Canceling state.

Parameters:

executionGraph - executionGraph to pass to the Canceling state

executionGraphHandler - executionGraphHandler to pass to the Canceling state

operatorCoordinatorHandler - operatorCoordinatorHandler to pass to the Canceling state

failureCollection - collection of failures that are propagated

goToRestarting

public void goToRestarting(ExecutionGraph executionGraph,
                           ExecutionGraphHandler executionGraphHandler,
                           OperatorCoordinatorHandler operatorCoordinatorHandler,
                           Duration backoffTime,
                           List<ExceptionHistoryEntry> failureCollection)

Description copied from interface: StateTransitions.ToRestarting

Transitions into the Restarting state.

Parameters:

executionGraph - executionGraph to pass to the Restarting state

executionGraphHandler - executionGraphHandler to pass to the Restarting state

operatorCoordinatorHandler - operatorCoordinatorHandler to pas to the Restarting state

backoffTime - backoffTime to wait before transitioning to the Restarting state

failureCollection - collection of failures that are propagated

goToFailing

public void goToFailing(ExecutionGraph executionGraph,
                        ExecutionGraphHandler executionGraphHandler,
                        OperatorCoordinatorHandler operatorCoordinatorHandler,
                        Throwable failureCause,
                        List<ExceptionHistoryEntry> failureCollection)

Description copied from interface: StateTransitions.ToFailing

Transitions into the Failing state.

Parameters:

executionGraph - executionGraph to pass to the Failing state

executionGraphHandler - executionGraphHandler to pass to the Failing state

operatorCoordinatorHandler - operatorCoordinatorHandler to pass to the Failing state

failureCause - failureCause describing why the job execution failed

failureCollection - collection of failures that are propagated

goToStopWithSavepoint

public CompletableFuture<String> goToStopWithSavepoint(ExecutionGraph executionGraph,
                                                       ExecutionGraphHandler executionGraphHandler,
                                                       OperatorCoordinatorHandler operatorCoordinatorHandler,
                                                       CheckpointScheduling checkpointScheduling,
                                                       CompletableFuture<String> savepointFuture,
                                                       List<ExceptionHistoryEntry> failureCollection)

Description copied from interface: StateTransitions.ToStopWithSavepoint

Transitions into the StopWithSavepoint state.

Parameters:

executionGraph - executionGraph to pass to the StopWithSavepoint state

executionGraphHandler - executionGraphHandler to pass to the StopWithSavepoint state

operatorCoordinatorHandler - operatorCoordinatorHandler to pass to the StopWithSavepoint state

savepointFuture - Future for the savepoint to complete.

failureCollection - collection of failures that are propagated

Returns:

Location of the savepoint.

goToFinished

public void goToFinished(ArchivedExecutionGraph archivedExecutionGraph)

Description copied from interface: StateTransitions.ToFinished

Transitions into the Finished state.

Parameters:

archivedExecutionGraph - archivedExecutionGraph which is passed to the Finished state

goToCreatingExecutionGraph

public void goToCreatingExecutionGraph(@Nullable
                                       ExecutionGraph previousExecutionGraph)

Description copied from interface: StateTransitions.ToCreatingExecutionGraph

Transitions into the CreatingExecutionGraph state.

tryToAssignSlots

public org.apache.flink.runtime.scheduler.adaptive.CreatingExecutionGraph.AssignmentResult tryToAssignSlots(org.apache.flink.runtime.scheduler.adaptive.CreatingExecutionGraph.ExecutionGraphWithVertexParallelism executionGraphWithVertexParallelism)

Try to assign slots to the created ExecutionGraph. If it is possible, then this method returns a successful AssignmentResult which contains the assigned ExecutionGraph. If not, then the assignment result is a failure.

Parameters:

executionGraphWithVertexParallelism - executionGraphWithVertexParallelism to assign slots to resources

Returns:

AssignmentResult representing the result of the assignment

shouldRescale

public boolean shouldRescale(ExecutionGraph executionGraph,
                             boolean forceRescale)

In regular mode, rescale the job if added resource meets JobManagerOptions.MIN_PARALLELISM_INCREASE. In force mode rescale if the parallelism has changed.

Parameters:

executionGraph - executionGraph for making the scaling decision.

forceRescale - should we force rescaling

Returns:

true, if we should rescale

onFinished

public void onFinished(ArchivedExecutionGraph archivedExecutionGraph)

Callback which is called when the execution reaches the Finished state.

Parameters:

archivedExecutionGraph - archivedExecutionGraph represents the final state of the job execution

howToHandleFailure

public org.apache.flink.runtime.scheduler.adaptive.FailureResult howToHandleFailure(Throwable failure,
                                                                                    CompletableFuture<Map<String,String>> failureLabels)

Asks how to handle the failure.

Parameters:

failure - failure describing the failure cause

failureLabels - future of labels from error classification.

Returns:

FailureResult which describes how to handle the failure

getIOExecutor

public Executor getIOExecutor()

Gets the I/O executor.

Returns:

the I/O executor

getMainThreadExecutor

public ComponentMainThreadExecutor getMainThreadExecutor()

Gets the main thread executor.

Returns:

the main thread executor

getMetricGroup

public JobManagerJobMetricGroup getMetricGroup()

Gets the JobManagerJobMetricGroup.

Returns:

the metric group

isState

public boolean isState(org.apache.flink.runtime.scheduler.adaptive.State expectedState)

Checks whether the current state is the expected state.

Parameters:

expectedState - expectedState is the expected state

Returns:

true if the current state equals the expected state; otherwise false

runIfState

public void runIfState(org.apache.flink.runtime.scheduler.adaptive.State expectedState,
                       Runnable action)

Run the given action if the current state equals the expected state.

Parameters:

expectedState - expectedState is the expected state

action - action to run if the current state equals the expected state

runIfState

public ScheduledFuture<?> runIfState(org.apache.flink.runtime.scheduler.adaptive.State expectedState,
                                     Runnable action,
                                     Duration delay)

Runs the given action after a delay if the state at this time equals the expected state.

Parameters:

expectedState - expectedState describes the required state at the time of running the action

action - action to run if the expected state equals the actual state

delay - delay after which to run the action

Returns:

a ScheduledFuture representing pending completion of the task