Package org.apache.flink.connector.base.source.reader

Class SourceReaderBase<E,T,SplitT extends org.apache.flink.api.connector.source.SourceSplit,SplitStateT>

java.lang.Object

org.apache.flink.connector.base.source.reader.SourceReaderBase<E,T,SplitT,SplitStateT>

Type Parameters:

E - The rich element type that contains information for split state update or timestamp extraction.

T - The final element type to emit.

SplitT - the immutable split type.

SplitStateT - the mutable type of split state.

All Implemented Interfaces:

AutoCloseable, org.apache.flink.api.common.state.CheckpointListener, org.apache.flink.api.connector.source.SourceReader<T,SplitT>

Direct Known Subclasses:

SingleThreadMultiplexSourceReaderBase

@PublicEvolving
public abstract class SourceReaderBase<E,T,SplitT extends org.apache.flink.api.connector.source.SourceSplit,SplitStateT>
extends Object
implements org.apache.flink.api.connector.source.SourceReader<T,SplitT>

An abstract implementation of SourceReader which provides some synchronization between the mail box main thread and the SourceReader internal threads. This class allows user to just provide a SplitReader and snapshot the split state.

This implementation provides the following metrics out of the box:

• OperatorIOMetricGroup.getNumRecordsInCounter()

Field Summary

Fields

Modifier and Type	Field	Description
protected org.apache.flink.configuration.Configuration	config	The raw configurations that may be used by subclasses.
protected org.apache.flink.api.connector.source.SourceReaderContext	context	The context of this source reader.
protected RecordEvaluator<T>	eofRecordEvaluator
protected SourceReaderOptions	options	The configuration for the reader.
protected RecordEmitter<E,T,SplitStateT>	recordEmitter	The record emitter to handle the records read by the SplitReaders.
protected SplitFetcherManager<E,SplitT>	splitFetcherManager	The split fetcher manager to run split fetchers.

Constructor Summary

Constructors

Constructor	Description
SourceReaderBase(SplitFetcherManager<E,SplitT> splitFetcherManager, RecordEmitter<E,T,SplitStateT> recordEmitter, org.apache.flink.configuration.Configuration config, org.apache.flink.api.connector.source.SourceReaderContext context)	The primary constructor for the source reader.
SourceReaderBase(SplitFetcherManager<E,SplitT> splitFetcherManager, RecordEmitter<E,T,SplitStateT> recordEmitter, RecordEvaluator<T> eofRecordEvaluator, org.apache.flink.configuration.Configuration config, org.apache.flink.api.connector.source.SourceReaderContext context)
SourceReaderBase(FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, SplitFetcherManager<E,SplitT> splitFetcherManager, RecordEmitter<E,T,SplitStateT> recordEmitter, org.apache.flink.configuration.Configuration config, org.apache.flink.api.connector.source.SourceReaderContext context)	Deprecated. Please use SourceReaderBase(SplitFetcherManager, RecordEmitter, Configuration, SourceReaderContext) instead.
SourceReaderBase(FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, SplitFetcherManager<E,SplitT> splitFetcherManager, RecordEmitter<E,T,SplitStateT> recordEmitter, RecordEvaluator<T> eofRecordEvaluator, org.apache.flink.configuration.Configuration config, org.apache.flink.api.connector.source.SourceReaderContext context)	Deprecated. Please use SourceReaderBase(SplitFetcherManager, RecordEmitter, RecordEvaluator, Configuration, SourceReaderContext) instead.

Method Summary

Modifier and Type	Method	Description
void	addSplits(List<SplitT> splits)
void	close()
int	getNumberOfCurrentlyAssignedSplits()	Gets the number of splits the reads has currently assigned.
void	handleSourceEvents(org.apache.flink.api.connector.source.SourceEvent sourceEvent)
protected abstract SplitStateT	initializedState(SplitT split)	When new splits are added to the reader.
CompletableFuture<Void>	isAvailable()
void	notifyNoMoreSplits()
protected abstract void	onSplitFinished(Map<String,SplitStateT> finishedSplitIds)	Handles the finished splits to clean the state if needed.
void	pauseOrResumeSplits(Collection<String> splitsToPause, Collection<String> splitsToResume)
org.apache.flink.core.io.InputStatus	pollNext(org.apache.flink.api.connector.source.ReaderOutput<T> output)
List<SplitT>	snapshotState(long checkpointId)
void	start()
protected abstract SplitT	toSplitType(String splitId, SplitStateT splitState)	Convert a mutable SplitStateT to immutable SplitT.

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.common.state.CheckpointListener

notifyCheckpointAborted

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

notifyCheckpointComplete

Field Detail

recordEmitter

protected final RecordEmitter<E,T,SplitStateT> recordEmitter

The record emitter to handle the records read by the SplitReaders.

splitFetcherManager

protected final SplitFetcherManager<E,SplitT extends org.apache.flink.api.connector.source.SourceSplit> splitFetcherManager

The split fetcher manager to run split fetchers.

options

protected final SourceReaderOptions options

The configuration for the reader.

config

protected final org.apache.flink.configuration.Configuration config

The raw configurations that may be used by subclasses.

context

protected org.apache.flink.api.connector.source.SourceReaderContext context

The context of this source reader.

eofRecordEvaluator

@Nullable
protected final RecordEvaluator<T> eofRecordEvaluator

Constructor Detail

SourceReaderBase

@Deprecated
public SourceReaderBase(FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
                        SplitFetcherManager<E,SplitT> splitFetcherManager,
                        RecordEmitter<E,T,SplitStateT> recordEmitter,
                        org.apache.flink.configuration.Configuration config,
                        org.apache.flink.api.connector.source.SourceReaderContext context)

Deprecated.

Please use SourceReaderBase(SplitFetcherManager, RecordEmitter, Configuration, SourceReaderContext) instead.

SourceReaderBase

@Deprecated
public SourceReaderBase(FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
                        SplitFetcherManager<E,SplitT> splitFetcherManager,
                        RecordEmitter<E,T,SplitStateT> recordEmitter,
                        @Nullable
                        RecordEvaluator<T> eofRecordEvaluator,
                        org.apache.flink.configuration.Configuration config,
                        org.apache.flink.api.connector.source.SourceReaderContext context)

Deprecated.

Please use SourceReaderBase(SplitFetcherManager, RecordEmitter, RecordEvaluator, Configuration, SourceReaderContext) instead.

SourceReaderBase

public SourceReaderBase(SplitFetcherManager<E,SplitT> splitFetcherManager,
                        RecordEmitter<E,T,SplitStateT> recordEmitter,
                        org.apache.flink.configuration.Configuration config,
                        org.apache.flink.api.connector.source.SourceReaderContext context)

The primary constructor for the source reader.

The reader will use a handover queue sized as configured via SourceReaderOptions.ELEMENT_QUEUE_CAPACITY.

SourceReaderBase

public SourceReaderBase(SplitFetcherManager<E,SplitT> splitFetcherManager,
                        RecordEmitter<E,T,SplitStateT> recordEmitter,
                        @Nullable
                        RecordEvaluator<T> eofRecordEvaluator,
                        org.apache.flink.configuration.Configuration config,
                        org.apache.flink.api.connector.source.SourceReaderContext context)

Method Detail

start

public void start()

Specified by:

start in interface org.apache.flink.api.connector.source.SourceReader<E,T>

pollNext

public org.apache.flink.core.io.InputStatus pollNext(org.apache.flink.api.connector.source.ReaderOutput<T> output)
                                              throws Exception

Specified by:

pollNext in interface org.apache.flink.api.connector.source.SourceReader<E,T>

Throws:

Exception

isAvailable

public CompletableFuture<Void> isAvailable()

Specified by:

isAvailable in interface org.apache.flink.api.connector.source.SourceReader<E,T>

snapshotState

public List<SplitT> snapshotState(long checkpointId)

Specified by:

snapshotState in interface org.apache.flink.api.connector.source.SourceReader<E,T>

addSplits

public void addSplits(List<SplitT> splits)

Specified by:

addSplits in interface org.apache.flink.api.connector.source.SourceReader<E,T>

notifyNoMoreSplits

public void notifyNoMoreSplits()

Specified by:

notifyNoMoreSplits in interface org.apache.flink.api.connector.source.SourceReader<E,T>

handleSourceEvents

public void handleSourceEvents(org.apache.flink.api.connector.source.SourceEvent sourceEvent)

Specified by:

handleSourceEvents in interface org.apache.flink.api.connector.source.SourceReader<E,T>

pauseOrResumeSplits

public void pauseOrResumeSplits(Collection<String> splitsToPause,
                                Collection<String> splitsToResume)

Specified by:

pauseOrResumeSplits in interface org.apache.flink.api.connector.source.SourceReader<E,T>

close

public void close()
           throws Exception

Specified by:

close in interface AutoCloseable

Throws:

Exception

getNumberOfCurrentlyAssignedSplits

public int getNumberOfCurrentlyAssignedSplits()

Gets the number of splits the reads has currently assigned.

These are the splits that have been added via addSplits(List) and have not yet been finished by returning them from the SplitReader.fetch() as part of RecordsWithSplitIds.finishedSplits().

onSplitFinished

protected abstract void onSplitFinished(Map<String,SplitStateT> finishedSplitIds)

Handles the finished splits to clean the state if needed.

initializedState

protected abstract SplitStateT initializedState(SplitT split)

When new splits are added to the reader. The initialize the state of the new splits.

Parameters:

split - a newly added split.

toSplitType

protected abstract SplitT toSplitType(String splitId,
                                      SplitStateT splitState)

Convert a mutable SplitStateT to immutable SplitT.

Parameters:

splitState - splitState.

Returns:

an immutable Split state.