Package org.apache.flink.runtime.asyncprocessing

Class AsyncExecutionController<K>

java.lang.Object

org.apache.flink.runtime.asyncprocessing.AsyncExecutionController<K>

Type Parameters:

K - the type of the key

All Implemented Interfaces:

Closeable, AutoCloseable, StateRequestHandler

public class AsyncExecutionController<K>
extends Object
implements StateRequestHandler, Closeable

The Async Execution Controller (AEC) receives processing requests from operators, and put them into execution according to some strategies.

It is responsible for:

Preserving the sequence of elements bearing the same key by delaying subsequent requests until the processing of preceding ones is finalized. Tracking the in-flight data(records) and blocking the input if too much data in flight (back-pressure). It invokes MailboxExecutor.yield() to pause current operations, allowing for the execution of callbacks (mails in Mailbox).

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	AsyncExecutionController.SwitchContextListener<K>	A listener listens the key context switch.

Constructor Summary

Constructors

Constructor	Description
AsyncExecutionController(MailboxExecutor mailboxExecutor, StateFutureImpl.AsyncFrameworkExceptionHandler exceptionHandler, StateExecutor stateExecutor, int maxParallelism, int batchSize, long bufferTimeout, int maxInFlightRecords, AsyncExecutionController.SwitchContextListener<K> switchContextListener)

Method Summary

Modifier and Type	Method	Description
RecordContext<K>	buildContext(Object record, K key)	Build a new context based on record and key.
void	close()
void	drainInflightRecords(int targetNum)	A helper function to drain in-flight records util inFlightRecordNum within the limit of given targetNum.
int	getInFlightRecordNum()
StateExecutor	getStateExecutor()
<IN,OUT> InternalStateFuture<OUT>	handleRequest(State state, StateRequestType type, IN payload)	Submit a StateRequest to this AsyncExecutionController and trigger it if needed.
<IN,OUT> OUT	handleRequestSync(State state, StateRequestType type, IN payload)	Submit a StateRequest to this StateRequestHandler, and wait for the response synchronously.
void	processNonRecord(ThrowingRunnable<? extends Exception> action)
void	setCurrentContext(RecordContext<K> switchingContext)	Each time before a code segment (callback) is about to run in mailbox (task thread), this method should be called to switch a context in AEC.
<N> void	setCurrentNamespaceForState(InternalPartitionedState<N> state, N namespace)	Set current namespace for a state.
void	syncPointRequestWithCallback(ThrowingRunnable<Exception> callback)	A helper to request a StateRequestType.SYNC_POINT and run a callback if it finishes (once the record is not blocked).
void	triggerIfNeeded(boolean force)	Trigger a batch of requests.

Methods inherited from class java.lang.Object

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

Constructor Detail

AsyncExecutionController

public AsyncExecutionController(MailboxExecutor mailboxExecutor,
                                StateFutureImpl.AsyncFrameworkExceptionHandler exceptionHandler,
                                StateExecutor stateExecutor,
                                int maxParallelism,
                                int batchSize,
                                long bufferTimeout,
                                int maxInFlightRecords,
                                AsyncExecutionController.SwitchContextListener<K> switchContextListener)

Method Detail

buildContext

public RecordContext<K> buildContext(Object record,
                                     K key)

Build a new context based on record and key. Also wired with internal KeyAccountingUnit.

Parameters:

record - the given record.

key - the given key.

Returns:

the built record context.

setCurrentContext

public void setCurrentContext(RecordContext<K> switchingContext)

Each time before a code segment (callback) is about to run in mailbox (task thread), this method should be called to switch a context in AEC.

Parameters:

switchingContext - the context to switch.

handleRequest

public <IN,OUT> InternalStateFuture<OUT> handleRequest(@Nullable
                                                             State state,
                                                             StateRequestType type,
                                                             @Nullable
                                                             IN payload)

Submit a StateRequest to this AsyncExecutionController and trigger it if needed.

Specified by:

handleRequest in interface StateRequestHandler

Parameters:

state - the state to request. Could be null if the type is StateRequestType.SYNC_POINT.

type - the type of this request.

payload - the payload input for this request.

Returns:

the state future.

handleRequestSync

public <IN,OUT> OUT handleRequestSync(State state,
                                            StateRequestType type,
                                            @Nullable
                                            IN payload)

Description copied from interface: StateRequestHandler

Submit a StateRequest to this StateRequestHandler, and wait for the response synchronously.

Specified by:

handleRequestSync in interface StateRequestHandler

Parameters:

state - the state to request.

type - the type of this request.

payload - the payload input for this request.

Returns:

the state future.

setCurrentNamespaceForState

public <N> void setCurrentNamespaceForState(@Nonnull
                                            InternalPartitionedState<N> state,
                                            N namespace)

Description copied from interface: StateRequestHandler

Set current namespace for a state. See InternalPartitionedState.setCurrentNamespace(Object).

Specified by:

setCurrentNamespaceForState in interface StateRequestHandler

triggerIfNeeded

public void triggerIfNeeded(boolean force)

Trigger a batch of requests.

Parameters:

force - whether to trigger requests in force.

syncPointRequestWithCallback

public void syncPointRequestWithCallback(ThrowingRunnable<Exception> callback)

A helper to request a StateRequestType.SYNC_POINT and run a callback if it finishes (once the record is not blocked).

Parameters:

callback - the callback to run if it finishes (once the record is not blocked).

drainInflightRecords

public void drainInflightRecords(int targetNum)

A helper function to drain in-flight records util inFlightRecordNum within the limit of given targetNum.

Parameters:

targetNum - the target inFlightRecordNum to achieve.

processNonRecord

public void processNonRecord(ThrowingRunnable<? extends Exception> action)

getStateExecutor

@VisibleForTesting
public StateExecutor getStateExecutor()

getInFlightRecordNum

@VisibleForTesting
public int getInFlightRecordNum()

close

public void close()
           throws IOException

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Throws:

IOException