OUT
- OP
- @Internal public abstract class StreamTask<OUT,OP extends StreamOperator<OUT>> extends AbstractInvokable implements StatefulTask, AsyncExceptionHandler
StreamOperator
s which form
the Task's operator chain. Operators that are chained together execute synchronously in the
same thread and hence on the same stream partition. A common case for these chains
are successive map/flatmap/filter tasks.
The task chain contains one "head" operator and multiple chained operators. The StreamTask is specialized for the type of the head operator: one-input and two-input tasks, as well as for sources, iteration heads and iteration tails.
The Task class deals with the setup of the streams read by the head operator, and the streams produced by the operators at the ends of the operator chain. Note that the chain may fork and thus have multiple ends.
The life cycle of the task is set up as follows:
-- setInitialState -> provides state of all operators in the chain
-- invoke()
|
+----> Create basic utils (config, etc) and load the chain of operators
+----> operators.setup()
+----> task specific init()
+----> initialize-operator-states()
+----> open-operators()
+----> run()
+----> close-operators()
+----> dispose-operators()
+----> common cleanup
+----> task specific cleanup()
The StreamTask
has a lock object called lock
. All calls to methods on a
StreamOperator
must be synchronized on this lock object to ensure that no methods
are called concurrently.
Modifier and Type | Field and Description |
---|---|
protected OP |
headOperator
the head operator that consumes the input streams of this task.
|
protected OperatorChain<OUT,OP> |
operatorChain
The chain of operators executed by this task.
|
static ThreadGroup |
TRIGGER_THREAD_GROUP
The thread group that holds all trigger timer threads.
|
Constructor and Description |
---|
StreamTask() |
Modifier and Type | Method and Description |
---|---|
void |
abortCheckpointOnBarrier(long checkpointId,
Throwable cause)
Aborts a checkpoint as the result of receiving possibly some checkpoint barriers,
but at least one
CancelCheckpointMarker . |
void |
cancel()
This method is called when a task is canceled either as a result of a user abort or an execution failure.
|
protected abstract void |
cancelTask() |
protected abstract void |
cleanup() |
CheckpointStreamFactory |
createCheckpointStreamFactory(StreamOperator<?> operator)
This is only visible because
GenericWriteAheadSink uses the
checkpoint stream factory to write write-ahead logs. |
<K> AbstractKeyedStateBackend<K> |
createKeyedStateBackend(TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange) |
OperatorStateBackend |
createOperatorStateBackend(StreamOperator<?> op,
Collection<OperatorStateHandle> restoreStateHandles) |
CheckpointStreamFactory |
createSavepointStreamFactory(StreamOperator<?> operator,
String targetLocation) |
protected void |
finalize()
The finalize method shuts down the timer.
|
Map<String,Accumulator<?,?>> |
getAccumulatorMap() |
ExecutorService |
getAsyncOperationsThreadPool() |
CloseableRegistry |
getCancelables() |
Object |
getCheckpointLock()
Gets the lock object on which all operations that involve data and state mutation have to lock.
|
StreamConfig |
getConfiguration() |
String |
getName()
Gets the name of the task, in the form "taskname (2/5)".
|
ProcessingTimeService |
getProcessingTimeService()
Returns the
ProcessingTimeService responsible for telling the current
processing time and registering timers. |
StreamStatusMaintainer |
getStreamStatusMaintainer() |
void |
handleAsyncException(String message,
Throwable exception)
Handles an exception thrown by another thread (e.g.
|
protected abstract void |
init() |
void |
invoke()
Starts the execution.
|
boolean |
isCanceled() |
boolean |
isRunning() |
void |
notifyCheckpointComplete(long checkpointId)
Invoked when a checkpoint has been completed, i.e., when the checkpoint coordinator has received
the notification from all participating tasks.
|
protected abstract void |
run() |
void |
setInitialState(TaskStateHandles taskStateHandles)
Sets the initial state of the operator, upon recovery.
|
void |
setProcessingTimeService(ProcessingTimeService timeProvider)
Allows the user to specify his own
TimerServiceProvider . |
String |
toString() |
boolean |
triggerCheckpoint(CheckpointMetaData checkpointMetaData,
CheckpointOptions checkpointOptions)
This method is called to trigger a checkpoint, asynchronously by the checkpoint
coordinator.
|
void |
triggerCheckpointOnBarrier(CheckpointMetaData checkpointMetaData,
CheckpointOptions checkpointOptions,
CheckpointMetrics checkpointMetrics)
This method is called when a checkpoint is triggered as a result of receiving checkpoint
barriers on all input streams.
|
getCurrentNumberOfSubtasks, getEnvironment, getExecutionConfig, getIndexInSubtaskGroup, getJobConfiguration, getTaskConfiguration, getUserCodeClassLoader, setEnvironment
public static final ThreadGroup TRIGGER_THREAD_GROUP
protected OP extends StreamOperator<OUT> headOperator
protected OperatorChain<OUT,OP extends StreamOperator<OUT>> operatorChain
public void setProcessingTimeService(ProcessingTimeService timeProvider)
TimerServiceProvider
.
By default a DefaultTimerService
is going to be provided.
Changing it can be useful for testing processing time functionality, such as
WindowAssigners
and Triggers
.public final void invoke() throws Exception
AbstractInvokable
Must be overwritten by the concrete task implementation. This method is called by the task manager when the actual execution of the task starts.
All resources should be cleaned up when the method returns. Make sure
to guard the code with try-finally
blocks where necessary.
invoke
in class AbstractInvokable
Exception
- Tasks may forward their exceptions for the TaskManager to handle through failure/recovery.public final void cancel() throws Exception
AbstractInvokable
cancel
in class AbstractInvokable
Exception
- thrown if any exception occurs during the execution of the user codepublic final boolean isRunning()
public final boolean isCanceled()
protected void finalize() throws Throwable
This should not be relied upon! It will cause shutdown to happen much later than if manual shutdown is attempted, and cause threads to linger for longer than needed.
public String getName()
public Object getCheckpointLock()
public StreamConfig getConfiguration()
public Map<String,Accumulator<?,?>> getAccumulatorMap()
public StreamStatusMaintainer getStreamStatusMaintainer()
public void setInitialState(TaskStateHandles taskStateHandles)
StatefulTask
setInitialState
in interface StatefulTask
taskStateHandles
- All state handle for the task.public boolean triggerCheckpoint(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) throws Exception
StatefulTask
This method is called for tasks that start the checkpoints by injecting the initial barriers,
i.e., the source tasks. In contrast, checkpoints on downstream operators, which are the result of
receiving checkpoint barriers, invoke the #triggerCheckpointOnBarrier(CheckpointMetaData, CheckpointMetrics)
method.
triggerCheckpoint
in interface StatefulTask
checkpointMetaData
- Meta data for about this checkpointcheckpointOptions
- Options for performing this checkpointfalse
if the checkpoint can not be carried out, true
otherwiseException
public void triggerCheckpointOnBarrier(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, CheckpointMetrics checkpointMetrics) throws Exception
StatefulTask
triggerCheckpointOnBarrier
in interface StatefulTask
checkpointMetaData
- Meta data for about this checkpointcheckpointOptions
- Options for performing this checkpointcheckpointMetrics
- Metrics about this checkpointException
- Exceptions thrown as the result of triggering a checkpoint are forwarded.public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) throws Exception
StatefulTask
CancelCheckpointMarker
.
This requires implementing tasks to forward a
CancelCheckpointMarker
to their outputs.
abortCheckpointOnBarrier
in interface StatefulTask
checkpointId
- The ID of the checkpoint to be aborted.cause
- The reason why the checkpoint was aborted during alignmentException
public ExecutorService getAsyncOperationsThreadPool()
public void notifyCheckpointComplete(long checkpointId) throws Exception
StatefulTask
notifyCheckpointComplete
in interface StatefulTask
checkpointId
- The ID of the checkpoint that is complete..Exception
- The notification method may forward its exceptions.public OperatorStateBackend createOperatorStateBackend(StreamOperator<?> op, Collection<OperatorStateHandle> restoreStateHandles) throws Exception
Exception
public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(TypeSerializer<K> keySerializer, int numberOfKeyGroups, KeyGroupRange keyGroupRange) throws Exception
Exception
public CheckpointStreamFactory createCheckpointStreamFactory(StreamOperator<?> operator) throws IOException
GenericWriteAheadSink
uses the
checkpoint stream factory to write write-ahead logs. This should not be used for
anything else.IOException
public CheckpointStreamFactory createSavepointStreamFactory(StreamOperator<?> operator, String targetLocation) throws IOException
IOException
public ProcessingTimeService getProcessingTimeService()
ProcessingTimeService
responsible for telling the current
processing time and registering timers.public void handleAsyncException(String message, Throwable exception)
In more detail, it marks task execution failed for an external reason (a reason other than the task code itself throwing an exception). If the task is already in a terminal state (such as FINISHED, CANCELED, FAILED), or if the task is already canceling this does nothing. Otherwise it sets the state to FAILED, and, if the invokable code is running, starts an asynchronous thread that aborts that code.
This method never blocks.
handleAsyncException
in interface AsyncExceptionHandler
public CloseableRegistry getCancelables()
Copyright © 2014–2018 The Apache Software Foundation. All rights reserved.