org.apache.flink.runtime.executiongraph

Class Execution

java.lang.Object

org.apache.flink.runtime.executiongraph.Execution

All Implemented Interfaces:

Archiveable<ArchivedExecution>, AccessExecution, LogicalSlot.Payload

public class Execution
extends Object
implements AccessExecution, Archiveable<ArchivedExecution>, LogicalSlot.Payload

A single execution of a vertex. While an ExecutionVertex can be executed multiple times (for recovery, re-computation, re-configuration), this class tracks the state of a single execution of that vertex and the resources.

Lock free state transitions

In several points of the code, we need to deal with possible concurrent state changes and actions. For example, while the call to deploy a task (send it to the TaskManager) happens, the task gets cancelled.

We could lock the entire portion of the code (decision to deploy, deploy, set state to running) such that it is guaranteed that any "cancel command" will only pick up after deployment is done and that the "cancel command" call will never overtake the deploying call.

This blocks the threads big time, because the remote calls may take long. Depending of their locking behavior, it may even result in distributed deadlocks (unless carefully avoided). We therefore use atomic state updates and occasional double-checking to ensure that the state after a completed call is as expected, and trigger correcting actions if it is not. Many actions are also idempotent (like canceling).

Constructor Summary

Constructors

Constructor and Description
Execution(Executor executor, ExecutionVertex vertex, int attemptNumber, long globalModVersion, long startTimestamp, Time rpcTimeout) Creates a new Execution attempt.

Method Summary

Modifier and Type	Method and Description
ArchivedExecution	archive()
CompletableFuture<Collection<TaskManagerLocation>>	calculatePreferredLocations(LocationPreferenceConstraint locationPreferenceConstraint) Calculates the preferred locations based on the location preference constraint.
void	cancel()
void	deploy() Deploys the execution to the previously assigned resource.
void	fail(Throwable t) This method fails the vertex due to an external condition.
AllocationID	getAssignedAllocationID()
LogicalSlot	getAssignedResource()
TaskManagerLocation	getAssignedResourceLocation() Returns the TaskManagerLocation for this execution.
ExecutionAttemptID	getAttemptId() Returns the ExecutionAttemptID for this Execution.
int	getAttemptNumber() Returns the attempt number for this execution.
Throwable	getFailureCause()
String	getFailureCauseAsString() Returns the exception that caused the job to fail.
long	getGlobalModVersion() Gets the global modification version of the execution graph when this execution was created.
IOMetrics	getIOMetrics()
InputSplit	getNextInputSplit()
int	getParallelSubtaskIndex() Returns the subtask index of this execution.
CompletableFuture<?>	getReleaseFuture() Gets the release future which is completed once the execution reaches a terminal state and the assigned resource has been released.
Optional<ResultPartitionDeploymentDescriptor>	getResultPartitionDeploymentDescriptor(IntermediateResultPartitionID id)
ExecutionState	getState() Returns the current ExecutionState for this execution.
long	getStateTimestamp(ExecutionState state) Returns the timestamp for the given ExecutionState.
long[]	getStateTimestamps() Returns the timestamps for every ExecutionState.
CompletableFuture<TaskManagerLocation>	getTaskManagerLocationFuture()
JobManagerTaskRestore	getTaskRestore()
CompletableFuture<ExecutionState>	getTerminalStateFuture() Gets a future that completes once the task execution reaches a terminal state.
Map<String,Accumulator<?,?>>	getUserAccumulators()
StringifiedAccumulatorResult[]	getUserAccumulatorsStringified() Returns the user-defined accumulators as strings.
ExecutionVertex	getVertex()
String	getVertexWithAttempt()
boolean	isFinished()
void	notifyCheckpointAborted(long abortCheckpointId, long timestamp) Notify the task of this execution about a aborted checkpoint.
void	notifyCheckpointComplete(long checkpointId, long timestamp) Notify the task of this execution about a completed checkpoint.
CompletableFuture<Execution>	registerProducedPartitions(TaskManagerLocation location)
CompletableFuture<Execution>	registerProducedPartitions(TaskManagerLocation location, boolean sendScheduleOrUpdateConsumersMessage)
CompletableFuture<TaskBackPressureResponse>	requestBackPressure(int requestId, Time timeout) Request the back pressure ratio from the task of this execution.
CompletableFuture<Void>	scheduleForExecution()
CompletableFuture<Void>	scheduleForExecution(SlotProviderStrategy slotProviderStrategy, LocationPreferenceConstraint locationPreferenceConstraint, Set<AllocationID> allPreviousExecutionGraphAllocationIds) NOTE: This method only throws exceptions if it is in an illegal state to be scheduled, or if the tasks needs to be scheduled immediately and no resource is available.
CompletableFuture<Acknowledge>	sendOperatorEvent(OperatorID operatorId, SerializedValue<OperatorEvent> event) Sends the operator event to the Task on the Task Executor.
void	setAccumulators(Map<String,Accumulator<?,?>> userAccumulators) Update accumulators (discarded when the Execution has already been terminated).
void	setInitialState(JobManagerTaskRestore taskRestore) Sets the initial state for the execution.
CompletableFuture<?>	suspend()
String	toString()
void	transitionState(ExecutionState targetState)
void	triggerCheckpoint(long checkpointId, long timestamp, CheckpointOptions checkpointOptions) Trigger a new checkpoint on the task of this execution.
void	triggerSynchronousSavepoint(long checkpointId, long timestamp, CheckpointOptions checkpointOptions) Trigger a new checkpoint on the task of this execution.
boolean	tryAssignResource(LogicalSlot logicalSlot) Tries to assign the given slot to the execution.

Methods inherited from class java.lang.Object

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

Constructor Detail

Execution

public Execution(Executor executor,
                 ExecutionVertex vertex,
                 int attemptNumber,
                 long globalModVersion,
                 long startTimestamp,
                 Time rpcTimeout)

Creates a new Execution attempt.

Parameters:

executor - The executor used to dispatch callbacks from futures and asynchronous RPC calls.

vertex - The execution vertex to which this Execution belongs

attemptNumber - The execution attempt number.

globalModVersion - The global modification version of the execution graph when this execution was created

startTimestamp - The timestamp that marks the creation of this Execution

rpcTimeout - The rpcTimeout for RPC calls like deploy/cancel/stop.

Method Detail

getVertex

public ExecutionVertex getVertex()

getAttemptId

public ExecutionAttemptID getAttemptId()

Description copied from interface: AccessExecution

Returns the ExecutionAttemptID for this Execution.

Specified by:

getAttemptId in interface AccessExecution

Returns:

ExecutionAttemptID for this execution

getAttemptNumber

public int getAttemptNumber()

Description copied from interface: AccessExecution

Returns the attempt number for this execution.

Specified by:

getAttemptNumber in interface AccessExecution

Returns:

attempt number for this execution.

getState

public ExecutionState getState()

Description copied from interface: AccessExecution

Returns the current ExecutionState for this execution.

Specified by:

getState in interface AccessExecution

Returns:

execution state for this execution

getAssignedAllocationID

@Nullable
public AllocationID getAssignedAllocationID()

getGlobalModVersion

public long getGlobalModVersion()

Gets the global modification version of the execution graph when this execution was created.

This version is bumped in the ExecutionGraph whenever a global failover happens. It is used to resolve conflicts between concurrent modification by global and local failover actions.

getTaskManagerLocationFuture

public CompletableFuture<TaskManagerLocation> getTaskManagerLocationFuture()

getAssignedResource

public LogicalSlot getAssignedResource()

getResultPartitionDeploymentDescriptor

public Optional<ResultPartitionDeploymentDescriptor> getResultPartitionDeploymentDescriptor(IntermediateResultPartitionID id)

tryAssignResource

public boolean tryAssignResource(LogicalSlot logicalSlot)

Tries to assign the given slot to the execution. The assignment works only if the Execution is in state SCHEDULED. Returns true, if the resource could be assigned.

Parameters:

logicalSlot - to assign to this execution

Returns:

true if the slot could be assigned to the execution, otherwise false

getNextInputSplit

public InputSplit getNextInputSplit()

getAssignedResourceLocation

public TaskManagerLocation getAssignedResourceLocation()

Description copied from interface: AccessExecution

Returns the TaskManagerLocation for this execution.

Specified by:

getAssignedResourceLocation in interface AccessExecution

Returns:

taskmanager location for this execution.

getFailureCause

public Throwable getFailureCause()

getFailureCauseAsString

public String getFailureCauseAsString()

Description copied from interface: AccessExecution

Returns the exception that caused the job to fail. This is the first root exception that was not recoverable and triggered job failure.

Specified by:

getFailureCauseAsString in interface AccessExecution

Returns:

failure exception as a string, or "(null)"

getStateTimestamps

public long[] getStateTimestamps()

Description copied from interface: AccessExecution

Returns the timestamps for every ExecutionState.

Specified by:

getStateTimestamps in interface AccessExecution

Returns:

timestamps for each state

getStateTimestamp

public long getStateTimestamp(ExecutionState state)

Description copied from interface: AccessExecution

Returns the timestamp for the given ExecutionState.

Specified by:

getStateTimestamp in interface AccessExecution

Parameters:

state - state for which the timestamp should be returned

Returns:

timestamp for the given state

isFinished

public boolean isFinished()

getTaskRestore

@Nullable
public JobManagerTaskRestore getTaskRestore()

setInitialState

public void setInitialState(@Nullable
                            JobManagerTaskRestore taskRestore)

Sets the initial state for the execution. The serialized state is then shipped via the TaskDeploymentDescriptor to the TaskManagers.

Parameters:

taskRestore - information to restore the state

getTerminalStateFuture

public CompletableFuture<ExecutionState> getTerminalStateFuture()

Gets a future that completes once the task execution reaches a terminal state. The future will be completed with specific state that the execution reached. This future is always completed from the job master's main thread.

Specified by:

getTerminalStateFuture in interface LogicalSlot.Payload

Returns:

A future which is completed once the execution reaches a terminal state

getReleaseFuture

public CompletableFuture<?> getReleaseFuture()

Gets the release future which is completed once the execution reaches a terminal state and the assigned resource has been released. This future is always completed from the job master's main thread.

Returns:

A future which is completed once the assigned resource has been released

scheduleForExecution

public CompletableFuture<Void> scheduleForExecution()

scheduleForExecution

public CompletableFuture<Void> scheduleForExecution(SlotProviderStrategy slotProviderStrategy,
                                                    LocationPreferenceConstraint locationPreferenceConstraint,
                                                    @Nonnull
                                                    Set<AllocationID> allPreviousExecutionGraphAllocationIds)

NOTE: This method only throws exceptions if it is in an illegal state to be scheduled, or if the tasks needs to be scheduled immediately and no resource is available. If the task is accepted by the schedule, any error sets the vertex state to failed and triggers the recovery logic.

Parameters:

slotProviderStrategy - The slot provider strategy to use to allocate slot for this execution attempt.

locationPreferenceConstraint - constraint for the location preferences

allPreviousExecutionGraphAllocationIds - set with all previous allocation ids in the job graph. Can be empty if the allocation ids are not required for scheduling.

Returns:

Future which is completed once the Execution has been deployed

registerProducedPartitions

public CompletableFuture<Execution> registerProducedPartitions(TaskManagerLocation location)

registerProducedPartitions

public CompletableFuture<Execution> registerProducedPartitions(TaskManagerLocation location,
                                                               boolean sendScheduleOrUpdateConsumersMessage)

deploy

public void deploy()
            throws JobException

Deploys the execution to the previously assigned resource.

Throws:

JobException - if the execution cannot be deployed to the assigned resource

cancel

public void cancel()

suspend

public CompletableFuture<?> suspend()

fail

public void fail(Throwable t)

This method fails the vertex due to an external condition. The task will move to state FAILED. If the task was in state RUNNING or DEPLOYING before, it will send a cancel call to the TaskManager.

Specified by:

fail in interface LogicalSlot.Payload

Parameters:

t - The exception that caused the task to fail.

requestBackPressure

public CompletableFuture<TaskBackPressureResponse> requestBackPressure(int requestId,
                                                                       Time timeout)

Request the back pressure ratio from the task of this execution.

Parameters:

requestId - id of the request.

timeout - the request times out.

Returns:

A future of the task back pressure result.

notifyCheckpointComplete

public void notifyCheckpointComplete(long checkpointId,
                                     long timestamp)

Notify the task of this execution about a completed checkpoint.

Parameters:

checkpointId - of the completed checkpoint

timestamp - of the completed checkpoint

notifyCheckpointAborted

public void notifyCheckpointAborted(long abortCheckpointId,
                                    long timestamp)

Notify the task of this execution about a aborted checkpoint.

Parameters:

abortCheckpointId - of the subsumed checkpoint

timestamp - of the subsumed checkpoint

triggerCheckpoint

public void triggerCheckpoint(long checkpointId,
                              long timestamp,
                              CheckpointOptions checkpointOptions)

Trigger a new checkpoint on the task of this execution.

Parameters:

checkpointId - of th checkpoint to trigger

timestamp - of the checkpoint to trigger

checkpointOptions - of the checkpoint to trigger

triggerSynchronousSavepoint

public void triggerSynchronousSavepoint(long checkpointId,
                                        long timestamp,
                                        CheckpointOptions checkpointOptions)

Trigger a new checkpoint on the task of this execution.

Parameters:

checkpointId - of th checkpoint to trigger

timestamp - of the checkpoint to trigger

checkpointOptions - of the checkpoint to trigger

sendOperatorEvent

public CompletableFuture<Acknowledge> sendOperatorEvent(OperatorID operatorId,
                                                        SerializedValue<OperatorEvent> event)

Sends the operator event to the Task on the Task Executor.

Returns:

True, of the message was sent, false is the task is currently not running.

calculatePreferredLocations

@VisibleForTesting
public CompletableFuture<Collection<TaskManagerLocation>> calculatePreferredLocations(LocationPreferenceConstraint locationPreferenceConstraint)

Calculates the preferred locations based on the location preference constraint.

Parameters:

locationPreferenceConstraint - constraint for the location preference

Returns:

Future containing the collection of preferred locations. This might not be completed if not all inputs have been a resource assigned.

transitionState

public void transitionState(ExecutionState targetState)

getVertexWithAttempt

public String getVertexWithAttempt()

setAccumulators

public void setAccumulators(Map<String,Accumulator<?,?>> userAccumulators)

Update accumulators (discarded when the Execution has already been terminated).

Parameters:

userAccumulators - the user accumulators

getUserAccumulators

public Map<String,Accumulator<?,?>> getUserAccumulators()

getUserAccumulatorsStringified

public StringifiedAccumulatorResult[] getUserAccumulatorsStringified()

Description copied from interface: AccessExecution

Returns the user-defined accumulators as strings.

Specified by:

getUserAccumulatorsStringified in interface AccessExecution

Returns:

user-defined accumulators as strings.

getParallelSubtaskIndex

public int getParallelSubtaskIndex()

Description copied from interface: AccessExecution

Returns the subtask index of this execution.

Specified by:

getParallelSubtaskIndex in interface AccessExecution

Returns:

subtask index of this execution.

getIOMetrics

public IOMetrics getIOMetrics()

Specified by:

getIOMetrics in interface AccessExecution

toString

public String toString()

Overrides:

toString in class Object

archive

public ArchivedExecution archive()

Specified by:

archive in interface Archiveable<ArchivedExecution>