StreamTask (Flink : 1.13-SNAPSHOT API)

java.lang.Object
- org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable
- - org.apache.flink.streaming.runtime.tasks.StreamTask<OUT,OP>

Type Parameters:

OUT -

OP -

All Implemented Interfaces:

AsyncExceptionHandler

Direct Known Subclasses:

AbstractTwoInputStreamTask, MultipleInputStreamTask, OneInputStreamTask, SourceOperatorStreamTask, SourceStreamTask
```
@Internal
public abstract class StreamTask<OUT,OP extends StreamOperator<OUT>>
extends AbstractInvokable
implements AsyncExceptionHandler
```
Base class for all streaming tasks. A task is the unit of local processing that is deployed and executed by the TaskManagers. Each task runs one or more StreamOperators 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.

Field Summary

Fields
Modifier and Type	Field and Description
`protected CheckpointStorage`	`checkpointStorage` Our checkpoint storage.
`protected StreamConfig`	`configuration` The configuration of this streaming task.
`protected StreamInputProcessor`	`inputProcessor` The input processor.
`protected static org.slf4j.Logger`	`LOG` The logger used by the StreamTask and its subclasses.
`protected MailboxProcessor`	`mailboxProcessor`
`protected OP`	`mainOperator` the main operator that consumes the input streams of this task.
`protected OperatorChain<OUT,OP>`	`operatorChain` The chain of operators executed by this task.
`protected StateBackend`	`stateBackend` Our state backend.
`protected TimerService`	`systemTimerService` In contrast to `timerService` we should not register any user timers here.
`protected TimerService`	`timerService` The internal `TimerService` used to define the current processing time (default = `System.currentTimeMillis()`) and register timers for tasks to be executed in the future.
`static ThreadGroup`	`TRIGGER_THREAD_GROUP` The thread group that holds all trigger timer threads.

Constructor Summary

Constructors
Modifier	Constructor and Description
`protected`	`StreamTask(Environment env)` Constructor for initialization, possibly with initial state (recovery / savepoint / etc).
`protected`	`StreamTask(Environment env, TimerService timerService)` Constructor for initialization, possibly with initial state (recovery / savepoint / etc).
`protected`	`StreamTask(Environment environment, TimerService timerService, Thread.UncaughtExceptionHandler uncaughtExceptionHandler)`
`protected`	`StreamTask(Environment environment, TimerService timerService, Thread.UncaughtExceptionHandler uncaughtExceptionHandler, StreamTaskActionExecutor actionExecutor)` Constructor for initialization, possibly with initial state (recovery / savepoint / etc).
`protected`	`StreamTask(Environment environment, TimerService timerService, Thread.UncaughtExceptionHandler uncaughtExceptionHandler, StreamTaskActionExecutor actionExecutor, TaskMailbox mailbox)`

Method Summary

All Methods Static Methods Instance Methods Abstract Methods Concrete Methods
Modifier and Type	Method and Description
`void`	`abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)` Aborts a checkpoint as the result of receiving possibly some checkpoint barriers, but at least one `CancelCheckpointMarker`.
`protected void`	`advanceToEndOfEventTime()` Emits the `MAX_WATERMARK` so that all registered timers are fired.
`protected void`	`afterInvoke()`
`Future<Void>`	`cancel()` This method is called when a task is canceled either as a result of a user abort or an execution failure.
`protected void`	`cancelTask()`
`protected void`	`cleanup()`
`protected void`	`cleanUpInvoke()`
`static <OUT> RecordWriterDelegate<SerializationDelegate<StreamRecord<OUT>>>`	`createRecordWriterDelegate(StreamConfig configuration, Environment environment)`
`StreamTaskStateInitializer`	`createStreamTaskStateInitializer()`
`protected void`	`declineCheckpoint(long checkpointId)`
`void`	`dispatchOperatorEvent(OperatorID operator, SerializedValue<OperatorEvent> event)`
`protected void`	`finalize()` The finalize method shuts down the timer.
`protected void`	`finishTask()` Instructs the task to go through its normal termination routine, i.e.
`protected long`	`getAsyncCheckpointStartDelayNanos()`
`ExecutorService`	`getAsyncOperationsThreadPool()`
`CloseableRegistry`	`getCancelables()`
`CheckpointStorageWorkerView`	`getCheckpointStorage()`
`protected CompletableFuture<Void>`	`getCompletionFuture()`
`StreamConfig`	`getConfiguration()`
`MailboxExecutorFactory`	`getMailboxExecutorFactory()`
`String`	`getName()` Gets the name of the task, in the form "taskname (2/5)".
`ProcessingTimeServiceFactory`	`getProcessingTimeServiceFactory()`
`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`	`isFailing()`
`boolean`	`isMailboxLoopRunning()`
`boolean`	`isRunning()`
`boolean`	`isUsingNonBlockingInput()`
`Future<Void>`	`notifyCheckpointAbortAsync(long checkpointId)` Invoked when a checkpoint has been aborted, i.e., when the checkpoint coordinator has received a decline message from one task and try to abort the targeted checkpoint by notification.
`Future<Void>`	`notifyCheckpointCompleteAsync(long checkpointId)` Invoked when a checkpoint has been completed, i.e., when the checkpoint coordinator has received the notification from all participating tasks.
`protected void`	`processInput(MailboxDefaultAction.Controller controller)` This method implements the default action of the task (e.g.
`void`	`restore()` This method can be called before `AbstractInvokable.invoke()` to restore an invokable object for the last valid state, if it has it.
`void`	`runMailboxLoop()`
`boolean`	`runMailboxStep()`
`protected Counter`	`setupNumRecordsInCounter(StreamOperator streamOperator)`
`String`	`toString()`
`Future<Boolean>`	`triggerCheckpointAsync(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions)` This method is called to trigger a checkpoint, asynchronously by the checkpoint coordinator.
`void`	`triggerCheckpointOnBarrier(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, CheckpointMetricsBuilder checkpointMetrics)` This method is called when a checkpoint is triggered as a result of receiving checkpoint barriers on all input streams.

Methods inherited from class org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable
getCurrentNumberOfSubtasks, getEnvironment, getExecutionConfig, getIndexInSubtaskGroup, getJobConfiguration, getTaskConfiguration, getUserCodeClassLoader, setShouldInterruptOnCancel, shouldInterruptOnCancel

Methods inherited from class java.lang.Object
clone, equals, getClass, hashCode, notify, notifyAll, wait, wait, wait

- Field Detail
  - TRIGGER_THREAD_GROUP
```
public static final ThreadGroup TRIGGER_THREAD_GROUP
```
    The thread group that holds all trigger timer threads.
  - LOG
```
protected static final org.slf4j.Logger LOG
```
    The logger used by the StreamTask and its subclasses.
  - inputProcessor
```
@Nullable
protected StreamInputProcessor inputProcessor
```
    The input processor. Initialized in init() method.
  - mainOperator
```
protected OP extends StreamOperator<OUT> mainOperator
```
    the main operator that consumes the input streams of this task.
  - operatorChain
```
protected OperatorChain<OUT,OP extends StreamOperator<OUT>> operatorChain
```
    The chain of operators executed by this task.
  - configuration
```
protected final StreamConfig configuration
```
    The configuration of this streaming task.
  - stateBackend
```
protected final StateBackend stateBackend
```
    Our state backend. We use this to create a keyed state backend.
  - checkpointStorage
```
protected final CheckpointStorage checkpointStorage
```
    Our checkpoint storage. We use this to create checkpoint streams.
  - timerService
```
protected final TimerService timerService
```
    The internal TimerService used to define the current processing time (default = System.currentTimeMillis()) and register timers for tasks to be executed in the future.
  - systemTimerService
```
protected final TimerService systemTimerService
```
    In contrast to timerService we should not register any user timers here. It should be used only for system level timers.
  - mailboxProcessor
```
protected final MailboxProcessor mailboxProcessor
```
- Constructor Detail
  - StreamTask
```
protected StreamTask(Environment env)
              throws Exception
```
    Constructor for initialization, possibly with initial state (recovery / savepoint / etc).
    
    Parameters:
    
    env - The task environment for this task.
    
    Throws:
    
    Exception
  - StreamTask
```
protected StreamTask(Environment env,
                     @Nullable
                     TimerService timerService)
              throws Exception
```
    Constructor for initialization, possibly with initial state (recovery / savepoint / etc).
    
    Parameters:
    
    env - The task environment for this task.
    
    timerService - Optionally, a specific timer service to use.
    
    Throws:
    
    Exception
  - StreamTask
```
protected StreamTask(Environment environment,
                     @Nullable
                     TimerService timerService,
                     Thread.UncaughtExceptionHandler uncaughtExceptionHandler)
              throws Exception
```
    Throws:
    
    Exception
  - StreamTask
```
protected StreamTask(Environment environment,
                     @Nullable
                     TimerService timerService,
                     Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
                     StreamTaskActionExecutor actionExecutor)
              throws Exception
```
    Constructor for initialization, possibly with initial state (recovery / savepoint / etc).
    This constructor accepts a special TimerService. By default (and if null is passes for the timer service) a DefaultTimerService will be used.
    
    Parameters:
    
    environment - The task environment for this task.
    
    timerService - Optionally, a specific timer service to use.
    
    uncaughtExceptionHandler - to handle uncaught exceptions in the async operations thread pool
    
    actionExecutor - a mean to wrap all actions performed by this task thread. Currently, only SynchronizedActionExecutor can be used to preserve locking semantics.
    
    Throws:
    
    Exception
  - StreamTask
```
protected StreamTask(Environment environment,
                     @Nullable
                     TimerService timerService,
                     Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
                     StreamTaskActionExecutor actionExecutor,
                     TaskMailbox mailbox)
              throws Exception
```
    Throws:
    
    Exception
- Method Detail
  - init
```
protected abstract void init()
                      throws Exception
```
    Throws:
    
    Exception
  - cancelTask
```
protected void cancelTask()
                   throws Exception
```
    Throws:
    
    Exception
  - cleanup
```
protected void cleanup()
                throws Exception
```
    Throws:
    
    Exception
  - processInput
```
protected void processInput(MailboxDefaultAction.Controller controller)
                     throws Exception
```
    This method implements the default action of the task (e.g. processing one event from the input). Implementations should (in general) be non-blocking.
    
    Parameters:
    
    controller - controller object for collaborative interaction between the action and the stream task.
    
    Throws:
    
    Exception - on any problems in the action.
  - advanceToEndOfEventTime
```
protected void advanceToEndOfEventTime()
                                throws Exception
```
    Emits the MAX_WATERMARK so that all registered timers are fired.
    This is used by the source task when the job is TERMINATED. In the case, we want all the timers registered throughout the pipeline to fire and the related state (e.g. windows) to be flushed.
    For tasks other than the source task, this method does nothing.
    
    Throws:
    
    Exception
  - finishTask
```
protected void finishTask()
                   throws Exception
```
    Instructs the task to go through its normal termination routine, i.e. exit the run-loop and call StreamOperator.close() and StreamOperator.dispose() on its operators.
    This is used by the source task to get out of the run-loop when the job is stopped with a savepoint.
    For tasks other than the source task, this method does nothing.
    
    Throws:
    
    Exception
  - createStreamTaskStateInitializer
```
public StreamTaskStateInitializer createStreamTaskStateInitializer()
```
  - setupNumRecordsInCounter
```
protected Counter setupNumRecordsInCounter(StreamOperator streamOperator)
```
  - restore
```
public final void restore()
                   throws Exception
```
    Description copied from class: AbstractInvokable
    
    This method can be called before AbstractInvokable.invoke() to restore an invokable object for the last valid state, if it has it.
    Every implementation determinate what should be restored by itself. (nothing happens by default).
    
    Overrides:
    
    restore in class AbstractInvokable
    
    Throws:
    
    Exception - Tasks may forward their exceptions for the TaskManager to handle through failure/recovery.
  - invoke
```
public final void invoke()
                  throws Exception
```
    Description copied from class: AbstractInvokable
    
    Starts the execution.
    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.
    
    Specified by:
    
    invoke in class AbstractInvokable
    
    Throws:
    
    Exception - Tasks may forward their exceptions for the TaskManager to handle through failure/recovery.
  - runMailboxStep
```
@VisibleForTesting
public boolean runMailboxStep()
                                          throws Exception
```
    Throws:
    
    Exception
  - isMailboxLoopRunning
```
@VisibleForTesting
public boolean isMailboxLoopRunning()
```
  - runMailboxLoop
```
public void runMailboxLoop()
                    throws Exception
```
    Throws:
    
    Exception
  - afterInvoke
```
protected void afterInvoke()
                    throws Exception
```
    Throws:
    
    Exception
  - cleanUpInvoke
```
protected void cleanUpInvoke()
                      throws Exception
```
    Throws:
    
    Exception
  - getCompletionFuture
```
protected CompletableFuture<Void> getCompletionFuture()
```
  - cancel
```
public final Future<Void> cancel()
                          throws Exception
```
    Description copied from class: AbstractInvokable
    
    This method is called when a task is canceled either as a result of a user abort or an execution failure. It can be overwritten to respond to shut down the user code properly.
    
    Overrides:
    
    cancel in class AbstractInvokable
    
    Returns:
    
    a future that is completed when this AbstractInvokable is fully terminated. Note that it may never complete if the invokable is stuck.
    
    Throws:
    
    Exception - thrown if any exception occurs during the execution of the user code
  - getMailboxExecutorFactory
```
public MailboxExecutorFactory getMailboxExecutorFactory()
```
  - isRunning
```
public final boolean isRunning()
```
  - isCanceled
```
public final boolean isCanceled()
```
  - isFailing
```
public final boolean isFailing()
```
  - finalize
```
protected void finalize()
                 throws Throwable
```
    The finalize method shuts down the timer. This is a fail-safe shutdown, in case the original shutdown method was never called.
    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.
    
    Overrides:
    
    finalize in class Object
    
    Throws:
    
    Throwable
  - getName
```
public final String getName()
```
    Gets the name of the task, in the form "taskname (2/5)".
    
    Returns:
    
    The name of the task.
  - getCheckpointStorage
```
public CheckpointStorageWorkerView getCheckpointStorage()
```
  - getConfiguration
```
public StreamConfig getConfiguration()
```
  - getStreamStatusMaintainer
```
public StreamStatusMaintainer getStreamStatusMaintainer()
```
  - triggerCheckpointAsync
```
public Future<Boolean> triggerCheckpointAsync(CheckpointMetaData checkpointMetaData,
                                              CheckpointOptions checkpointOptions)
```
    Description copied from class: AbstractInvokable
    
    This method is called to trigger a checkpoint, asynchronously by the checkpoint coordinator.
    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 AbstractInvokable.triggerCheckpointOnBarrier(CheckpointMetaData, CheckpointOptions, CheckpointMetricsBuilder) method.
    
    Overrides:
    
    triggerCheckpointAsync in class AbstractInvokable
    
    Parameters:
    
    checkpointMetaData - Meta data for about this checkpoint
    
    checkpointOptions - Options for performing this checkpoint
    
    Returns:
    
    future with value of false if the checkpoint was not carried out, true otherwise
  - triggerCheckpointOnBarrier
```
public void triggerCheckpointOnBarrier(CheckpointMetaData checkpointMetaData,
                                       CheckpointOptions checkpointOptions,
                                       CheckpointMetricsBuilder checkpointMetrics)
                                throws IOException
```
    Description copied from class: AbstractInvokable
    
    This method is called when a checkpoint is triggered as a result of receiving checkpoint barriers on all input streams.
    
    Overrides:
    
    triggerCheckpointOnBarrier in class AbstractInvokable
    
    Parameters:
    
    checkpointMetaData - Meta data for about this checkpoint
    
    checkpointOptions - Options for performing this checkpoint
    
    checkpointMetrics - Metrics about this checkpoint
    
    Throws:
    
    IOException
  - abortCheckpointOnBarrier
```
public void abortCheckpointOnBarrier(long checkpointId,
                                     CheckpointException cause)
                              throws IOException
```
    Description copied from class: AbstractInvokable
    
    Aborts a checkpoint as the result of receiving possibly some checkpoint barriers, but at least one CancelCheckpointMarker.
    This requires implementing tasks to forward a CancelCheckpointMarker to their outputs.
    
    Overrides:
    
    abortCheckpointOnBarrier in class AbstractInvokable
    
    Parameters:
    
    checkpointId - The ID of the checkpoint to be aborted.
    
    cause - The reason why the checkpoint was aborted during alignment
    
    Throws:
    
    IOException
  - declineCheckpoint
```
protected void declineCheckpoint(long checkpointId)
```
  - getAsyncOperationsThreadPool
```
public final ExecutorService getAsyncOperationsThreadPool()
```
  - notifyCheckpointCompleteAsync
```
public Future<Void> notifyCheckpointCompleteAsync(long checkpointId)
```
    Description copied from class: AbstractInvokable
    
    Invoked when a checkpoint has been completed, i.e., when the checkpoint coordinator has received the notification from all participating tasks.
    
    Overrides:
    
    notifyCheckpointCompleteAsync in class AbstractInvokable
    
    Parameters:
    
    checkpointId - The ID of the checkpoint that is complete.
    
    Returns:
    
    future that completes when the notification has been processed by the task.
  - notifyCheckpointAbortAsync
```
public Future<Void> notifyCheckpointAbortAsync(long checkpointId)
```
    Description copied from class: AbstractInvokable
    
    Invoked when a checkpoint has been aborted, i.e., when the checkpoint coordinator has received a decline message from one task and try to abort the targeted checkpoint by notification.
    
    Overrides:
    
    notifyCheckpointAbortAsync in class AbstractInvokable
    
    Parameters:
    
    checkpointId - The ID of the checkpoint that is aborted.
    
    Returns:
    
    future that completes when the notification has been processed by the task.
  - dispatchOperatorEvent
```
public void dispatchOperatorEvent(OperatorID operator,
                                  SerializedValue<OperatorEvent> event)
                           throws FlinkException
```
    Overrides:
    
    dispatchOperatorEvent in class AbstractInvokable
    
    Throws:
    
    FlinkException
  - getProcessingTimeServiceFactory
```
public ProcessingTimeServiceFactory getProcessingTimeServiceFactory()
```
  - handleAsyncException
```
public void handleAsyncException(String message,
                                 Throwable exception)
```
    Handles an exception thrown by another thread (e.g. a TriggerTask), other than the one executing the main task by failing the task entirely.
    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.
    
    Specified by:
    
    handleAsyncException in interface AsyncExceptionHandler
  - toString
```
public String toString()
```
    Overrides:
    
    toString in class Object
  - getCancelables
```
public final CloseableRegistry getCancelables()
```
  - createRecordWriterDelegate
```
@VisibleForTesting
public static <OUT> RecordWriterDelegate<SerializationDelegate<StreamRecord<OUT>>> createRecordWriterDelegate(StreamConfig configuration,
                                                                                                                                 Environment environment)
```
  - getAsyncCheckpointStartDelayNanos
```
protected long getAsyncCheckpointStartDelayNanos()
```
  - isUsingNonBlockingInput
```
public boolean isUsingNonBlockingInput()
```
    Overrides:
    
    isUsingNonBlockingInput in class AbstractInvokable
    
    Returns:
    
    true if blocking input such as InputGate.getNext() is used (as opposed to InputGate.pollNext(). To be removed together with the DataSet API.

Back to Flink Website

Class StreamTask<OUT,OP extends StreamOperator<OUT>>

Field Summary

Constructor Summary

Method Summary

Methods inherited from class org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable

Methods inherited from class java.lang.Object

Field Detail

TRIGGER_THREAD_GROUP

LOG

inputProcessor

mainOperator

operatorChain

configuration

stateBackend

checkpointStorage

timerService

systemTimerService

mailboxProcessor

Constructor Detail

StreamTask

StreamTask

StreamTask

StreamTask

StreamTask

Method Detail

init

cancelTask

cleanup

processInput

advanceToEndOfEventTime

finishTask

createStreamTaskStateInitializer

setupNumRecordsInCounter

restore

invoke

runMailboxStep

isMailboxLoopRunning

runMailboxLoop

afterInvoke

cleanUpInvoke

getCompletionFuture

cancel

getMailboxExecutorFactory

isRunning

isCanceled

isFailing

finalize

getName

getCheckpointStorage

getConfiguration

getStreamStatusMaintainer

triggerCheckpointAsync

triggerCheckpointOnBarrier

abortCheckpointOnBarrier

declineCheckpoint

getAsyncOperationsThreadPool

notifyCheckpointCompleteAsync

notifyCheckpointAbortAsync

dispatchOperatorEvent

getProcessingTimeServiceFactory

handleAsyncException

toString

getCancelables

createRecordWriterDelegate

getAsyncCheckpointStartDelayNanos

isUsingNonBlockingInput

Back to Flink Website