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
CompletableFuture<Execution>	allocateAndAssignSlotForExecution(SlotProvider slotProvider, boolean queued, LocationPreferenceConstraint locationPreferenceConstraint, Time allocationTimeout) Allocates and assigns a slot obtained from the slot provider to the execution.
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()
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.
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	notifyCheckpointComplete(long checkpointId, long timestamp) Notify the task of this execution about a completed checkpoint.
CompletableFuture<StackTraceSampleResponse>	requestStackTraceSample(int sampleId, int numSamples, Time delayBetweenSamples, int maxStackTraceDepth, Time timeout) Request a stack trace sample from the task of this execution.
CompletableFuture<Void>	scheduleForExecution()
CompletableFuture<Void>	scheduleForExecution(SlotProvider slotProvider, boolean queued, LocationPreferenceConstraint locationPreferenceConstraint) 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.
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.
void	stop() Sends stop RPC call.
String	toString()
void	triggerCheckpoint(long checkpointId, long timestamp, CheckpointOptions checkpointOptions) Trigger a new checkpoint on the task of this 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()

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.

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.

Returns:

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

scheduleForExecution

public CompletableFuture<Void> scheduleForExecution()

scheduleForExecution

public CompletableFuture<Void> scheduleForExecution(SlotProvider slotProvider,
                                                    boolean queued,
                                                    LocationPreferenceConstraint locationPreferenceConstraint)

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:

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

queued - Flag to indicate whether the scheduler may queue this task if it cannot immediately deploy it.

locationPreferenceConstraint - constraint for the location preferences

Returns:

Future which is completed once the Execution has been deployed

allocateAndAssignSlotForExecution

public CompletableFuture<Execution> allocateAndAssignSlotForExecution(SlotProvider slotProvider,
                                                                      boolean queued,
                                                                      LocationPreferenceConstraint locationPreferenceConstraint,
                                                                      Time allocationTimeout)
                                                               throws IllegalExecutionStateException

Allocates and assigns a slot obtained from the slot provider to the execution.

Parameters:

slotProvider - to obtain a new slot from

queued - if the allocation can be queued

locationPreferenceConstraint - constraint for the location preferences

allocationTimeout - rpcTimeout for allocating a new slot

Returns:

Future which is completed with this execution once the slot has been assigned or with an exception if an error occurred.

Throws:

IllegalExecutionStateException - if this method has been called while not being in the CREATED state

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

stop

public void stop()

Sends stop RPC call.

cancel

public void cancel()

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.

requestStackTraceSample

public CompletableFuture<StackTraceSampleResponse> requestStackTraceSample(int sampleId,
                                                                           int numSamples,
                                                                           Time delayBetweenSamples,
                                                                           int maxStackTraceDepth,
                                                                           Time timeout)

Request a stack trace sample from the task of this execution.

Parameters:

sampleId - of the stack trace sample

numSamples - the sample should contain

delayBetweenSamples - to wait

maxStackTraceDepth - of the samples

timeout - until the request times out

Returns:

Future stack trace sample response

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

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

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.

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>