org.apache.flink.runtime.source.coordinator

Class SourceCoordinatorContext<SplitT extends SourceSplit>

java.lang.Object

org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext<SplitT>

Type Parameters:

SplitT - the type of the splits.

All Implemented Interfaces:

AutoCloseable, SplitEnumeratorContext<SplitT>, SupportsIntermediateNoMoreSplits

@Internal
public class SourceCoordinatorContext<SplitT extends SourceSplit>
extends Object
implements SplitEnumeratorContext<SplitT>, SupportsIntermediateNoMoreSplits, AutoCloseable

A context class for the OperatorCoordinator. Compared with SplitEnumeratorContext this class allows interaction with state and sending OperatorEvent to the SourceOperator while SplitEnumeratorContext only allows sending SourceEvent.

The context serves a few purposes:

• Information provider - The context provides necessary information to the enumerator for it to know what is the status of the source readers and their split assignments. These information allows the split enumerator to do the coordination.
• Action taker - The context also provides a few actions that the enumerator can take to carry out the coordination. So far there are two actions: 1) assign splits to the source readers. and 2) sens a custom SourceEvents to the source readers.
• Thread model enforcement - The context ensures that all the manipulations to the coordinator state are handled by the same thread.

Constructor Summary

Constructors

Constructor and Description
SourceCoordinatorContext(SourceCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory, int numWorkerThreads, OperatorCoordinator.Context operatorCoordinatorContext, SimpleVersionedSerializer<SplitT> splitSerializer, boolean supportsConcurrentExecutionAttempts)

Method Summary

Modifier and Type	Method and Description
void	assignSplits(SplitsAssignment<SplitT> assignment) Assign the splits.
<T> void	callAsync(Callable<T> callable, BiConsumer<T,Throwable> handler) Invoke the callable and handover the return value to the handler which will be executed by the source coordinator.
<T> void	callAsync(Callable<T> callable, BiConsumer<T,Throwable> handler, long initialDelay, long period) Invoke the given callable periodically and handover the return value to the handler which will be executed by the source coordinator.
void	close()
int	currentParallelism() Get the current parallelism of this Source.
TernaryBoolean	isBacklog() Returns whether the Source is processing backlog data.
SplitEnumeratorMetricGroup	metricGroup()
Map<Integer,ReaderInfo>	registeredReaders() Get the currently registered readers.
Map<Integer,Map<Integer,ReaderInfo>>	registeredReadersOfAttempts() Get the currently registered readers of all the subtask attempts.
void	runInCoordinatorThread(Runnable runnable) Invoke the given runnable in the source coordinator thread.
void	sendEventToSourceReader(int subtaskId, int attemptNumber, SourceEvent event) Send a source event to a source reader.
void	sendEventToSourceReader(int subtaskId, SourceEvent event) Send a source event to a source reader.
void	setIsProcessingBacklog(boolean isProcessingBacklog) Reports to JM whether this source is currently processing backlog.
void	signalIntermediateNoMoreSplits(int subtask) Signals a subtask that it will not receive split for current source, but it will receive split for next sources.
void	signalNoMoreSplits(int subtask) Signals a subtask that it will not receive any further split.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.api.connector.source.SplitEnumeratorContext

assignSplit

Constructor Detail

SourceCoordinatorContext

public SourceCoordinatorContext(SourceCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory,
                                int numWorkerThreads,
                                OperatorCoordinator.Context operatorCoordinatorContext,
                                SimpleVersionedSerializer<SplitT> splitSerializer,
                                boolean supportsConcurrentExecutionAttempts)

Method Detail

metricGroup

public SplitEnumeratorMetricGroup metricGroup()

Specified by:

metricGroup in interface SplitEnumeratorContext<SplitT extends SourceSplit>

sendEventToSourceReader

public void sendEventToSourceReader(int subtaskId,
                                    SourceEvent event)

Description copied from interface: SplitEnumeratorContext

Send a source event to a source reader. The source reader is identified by its subtask id.

Specified by:

sendEventToSourceReader in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

subtaskId - the subtask id of the source reader to send this event to.

event - the source event to send.

sendEventToSourceReader

public void sendEventToSourceReader(int subtaskId,
                                    int attemptNumber,
                                    SourceEvent event)

Description copied from interface: SplitEnumeratorContext

Send a source event to a source reader. The source reader is identified by its subtask id and attempt number. It is similar to SplitEnumeratorContext.sendEventToSourceReader(int, SourceEvent) but it is aware of the subtask execution attempt to send this event to.

The SplitEnumerator must invoke this method instead of SplitEnumeratorContext.sendEventToSourceReader(int, SourceEvent) if it is used in cases that a subtask can have multiple concurrent execution attempts, e.g. if speculative execution is enabled. Otherwise an error will be thrown when the split enumerator tries to send a custom source event.

Specified by:

sendEventToSourceReader in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

subtaskId - the subtask id of the source reader to send this event to.

attemptNumber - the attempt number of the source reader to send this event to.

event - the source event to send.

currentParallelism

public int currentParallelism()

Description copied from interface: SplitEnumeratorContext

Get the current parallelism of this Source. Note that due to auto-scaling, the parallelism may change over time. Therefore the SplitEnumerator should not cache the return value of this method, but always invoke this method to get the latest parallelism.

Specified by:

currentParallelism in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Returns:

the parallelism of the Source.

registeredReaders

public Map<Integer,ReaderInfo> registeredReaders()

Description copied from interface: SplitEnumeratorContext

Get the currently registered readers. The mapping is from subtask id to the reader info.

Note that if a subtask has multiple concurrent attempts, the map will contain the earliest attempt of that subtask. This is for compatibility purpose. It's recommended to use SplitEnumeratorContext.registeredReadersOfAttempts() instead.

Specified by:

registeredReaders in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Returns:

the currently registered readers.

registeredReadersOfAttempts

public Map<Integer,Map<Integer,ReaderInfo>> registeredReadersOfAttempts()

Description copied from interface: SplitEnumeratorContext

Get the currently registered readers of all the subtask attempts. The mapping is from subtask id to a map which maps an attempt to its reader info.

Specified by:

registeredReadersOfAttempts in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Returns:

the currently registered readers.

assignSplits

public void assignSplits(SplitsAssignment<SplitT> assignment)

Description copied from interface: SplitEnumeratorContext

Assign the splits.

Specified by:

assignSplits in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

assignment - the new split assignments to add.

signalNoMoreSplits

public void signalNoMoreSplits(int subtask)

Description copied from interface: SplitEnumeratorContext

Signals a subtask that it will not receive any further split.

Specified by:

signalNoMoreSplits in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

subtask - The index of the operator's parallel subtask that shall be signaled it will not receive any further split.

signalIntermediateNoMoreSplits

public void signalIntermediateNoMoreSplits(int subtask)

Description copied from interface: SupportsIntermediateNoMoreSplits

Signals a subtask that it will not receive split for current source, but it will receive split for next sources. A common scenario is HybridSource. This indicates that task not truly read finished.

Specified by:

signalIntermediateNoMoreSplits in interface SupportsIntermediateNoMoreSplits

Parameters:

subtask - The index of the operator's parallel subtask that shall be signaled it will receive splits later.

callAsync

public <T> void callAsync(Callable<T> callable,
                          BiConsumer<T,Throwable> handler,
                          long initialDelay,
                          long period)

Description copied from interface: SplitEnumeratorContext

Invoke the given callable periodically and handover the return value to the handler which will be executed by the source coordinator. When this method is invoked multiple times, The Callables may be executed in a thread pool concurrently.

It is important to make sure that the callable does not modify any shared state, especially the states that will be a part of the SplitEnumerator.snapshotState(long). Otherwise, there might be unexpected behavior.

Note that an exception thrown from the handler would result in failing the job.

Specified by:

callAsync in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

callable - the callable to call.

handler - a handler that handles the return value of or the exception thrown from the callable.

initialDelay - the initial delay of calling the callable, in milliseconds.

period - the period between two invocations of the callable, in milliseconds.

callAsync

public <T> void callAsync(Callable<T> callable,
                          BiConsumer<T,Throwable> handler)

Description copied from interface: SplitEnumeratorContext

Invoke the callable and handover the return value to the handler which will be executed by the source coordinator. When this method is invoked multiple times, The Callable s may be executed in a thread pool concurrently.

It is important to make sure that the callable does not modify any shared state, especially the states that will be a part of the SplitEnumerator.snapshotState(long). Otherwise, there might be unexpected behavior.

Note that an exception thrown from the handler would result in failing the job.

Specified by:

callAsync in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

callable - a callable to call.

handler - a handler that handles the return value of or the exception thrown from the callable.

runInCoordinatorThread

public void runInCoordinatorThread(Runnable runnable)

Invoke the given runnable in the source coordinator thread.

This can be useful when the enumerator needs to execute some action (like assignSplits) triggered by some external events. E.g., Watermark from another source advanced and this source now be able to assign splits to awaiting readers. The trigger can be initiated from the coordinator thread of the other source. Instead of using lock for thread safety, this API allows to run such externally triggered action in the coordinator thread. Hence, we can ensure all enumerator actions are serialized in the single coordinator thread.

It is important that the runnable does not block. If the runnable throws an Exception, the corresponding job is failed.

Specified by:

runInCoordinatorThread in interface SplitEnumeratorContext<SplitT extends SourceSplit>

Parameters:

runnable - a runnable to execute

close

public void close()
           throws InterruptedException

Specified by:

close in interface AutoCloseable

Throws:

InterruptedException

setIsProcessingBacklog

public void setIsProcessingBacklog(boolean isProcessingBacklog)

Description copied from interface: SplitEnumeratorContext

Reports to JM whether this source is currently processing backlog.

When source is processing backlog, it means the records being emitted by this source is already stale and there is no processing latency requirement for these records. This allows downstream operators to optimize throughput instead of reducing latency for intermediate results.

If no API has been explicitly invoked to specify the backlog status of a source, the source is considered to have isProcessingBacklog=false by default.

Specified by:

setIsProcessingBacklog in interface SplitEnumeratorContext<SplitT extends SourceSplit>

isBacklog

public TernaryBoolean isBacklog()

Returns whether the Source is processing backlog data. UNDEFINED is returned if it is not set by the setIsProcessingBacklog(boolean) method.