Package org.apache.beam.runners.flink.translation.wrappers.streaming.io.source.unbounded

Class FlinkUnboundedSourceReader<T>

java.lang.Object

org.apache.beam.runners.flink.translation.wrappers.streaming.io.source.FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

org.apache.beam.runners.flink.translation.wrappers.streaming.io.source.unbounded.FlinkUnboundedSourceReader<T>

Type Parameters:

T - the output element type of the encapsulated Beam UnboundedReader.

All Implemented Interfaces:

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

public class FlinkUnboundedSourceReader<T>
extends FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

A Flink SourceReader implementation that reads from the assigned FlinkSourceSplits by using Beam UnboundedReaders.

This reader consumes all the assigned source splits concurrently.

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.beam.runners.flink.translation.wrappers.streaming.io.source.FlinkSourceReaderBase

FlinkSourceReaderBase.ReaderAndOutput

Field Summary

Fields

Modifier and Type	Field	Description
protected static final String	PENDING_BYTES_METRIC_NAME

Fields inherited from class org.apache.beam.runners.flink.translation.wrappers.streaming.io.source.FlinkSourceReaderBase

AVAILABLE_NOW, context, idleTimeoutMs, invocationUtil, numRecordsInCounter, pipelineOptions, timestampExtractor

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	FlinkUnboundedSourceReader(String stepName, org.apache.flink.api.connector.source.SourceReaderContext context, PipelineOptions pipelineOptions, ScheduledExecutorService executor, @Nullable Function<WindowedValue<ValueWithRecordId<T>>,Long> timestampExtractor)
	FlinkUnboundedSourceReader(String stepName, org.apache.flink.api.connector.source.SourceReaderContext context, PipelineOptions pipelineOptions, @Nullable Function<WindowedValue<ValueWithRecordId<T>>,Long> timestampExtractor)

Method Summary

Modifier and Type	Method	Description
protected void	addSplitsToUnfinishedForCheckpoint(long checkpointId, List<FlinkSourceSplit<T>> flinkSourceSplits)	To be overridden in unbounded reader.
protected Source.Reader<T>	createReader(FlinkSourceSplit<T> sourceSplit)	Create Source.Reader for given FlinkSourceSplit.
protected FlinkSourceSplit<T>	getReaderCheckpoint(int splitId, FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>.ReaderAndOutput readerAndOutput)	Create FlinkSourceSplit for given splitId.
protected CompletableFuture<Void>	isAvailableForAliveReaders()	Check whether there are data available from alive readers.
void	notifyCheckpointComplete(long checkpointId)
org.apache.flink.core.io.InputStatus	pollNext(org.apache.flink.api.connector.source.ReaderOutput<WindowedValue<ValueWithRecordId<T>>> output)
void	start()

Methods inherited from class org.apache.beam.runners.flink.translation.wrappers.streaming.io.source.FlinkSourceReaderBase

addSplits, allReaders, checkExceptionAndMaybeThrow, checkIdleTimeoutAndMaybeStartCountdown, close, createAndTrackNextReader, execute, finishSplit, hasException, ignoreReturnValue, isAvailable, noMoreSplits, notifyNoMoreSplits, recordException, scheduleTask, scheduleTaskAtFixedRate, snapshotState, sourceSplits

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

handleSourceEvents, pauseOrResumeSplits

Field Details

PENDING_BYTES_METRIC_NAME

protected static final String PENDING_BYTES_METRIC_NAME

See Also:

• Constant Field Values

Constructor Details

FlinkUnboundedSourceReader

public FlinkUnboundedSourceReader(String stepName,
 org.apache.flink.api.connector.source.SourceReaderContext context,
 PipelineOptions pipelineOptions,
 @Nullable Function<WindowedValue<ValueWithRecordId<T>>,Long> timestampExtractor)

FlinkUnboundedSourceReader

protected FlinkUnboundedSourceReader(String stepName,
 org.apache.flink.api.connector.source.SourceReaderContext context,
 PipelineOptions pipelineOptions,
 ScheduledExecutorService executor,
 @Nullable Function<WindowedValue<ValueWithRecordId<T>>,Long> timestampExtractor)

Method Details

addSplitsToUnfinishedForCheckpoint

protected void addSplitsToUnfinishedForCheckpoint(long checkpointId,
 List<FlinkSourceSplit<T>> flinkSourceSplits)

Description copied from class: FlinkSourceReaderBase

To be overridden in unbounded reader. Notify the reader of created splits that will be part of checkpoint. Will be processed during notifyCheckpointComplete to finalize the associated CheckpointMarks.

Overrides:

addSplitsToUnfinishedForCheckpoint in class FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

notifyCheckpointComplete

public void notifyCheckpointComplete(long checkpointId)
                              throws Exception

Throws:

Exception

start

public void start()

Specified by:

start in interface org.apache.flink.api.connector.source.SourceReader<WindowedValue<ValueWithRecordId<T>>,FlinkSourceSplit<T>>

Overrides:

start in class FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

pollNext

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

Throws:

Exception

isAvailableForAliveReaders

protected CompletableFuture<Void> isAvailableForAliveReaders()

Check whether there are data available from alive readers. If not, set a future and wait for the periodically running wake-up task to complete that future when the check interval passes. This method is only called by the main thread, which is the only thread writing to the future ref. Note that for UnboundedSource, because the splits never finishes, there are always alive readers after the first split assigment. Hence, the return value of FlinkSourceReaderBase.isAvailable() will effectively be determined by this method after the first split assignment.

Specified by:

isAvailableForAliveReaders in class FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

getReaderCheckpoint

protected FlinkSourceSplit<T> getReaderCheckpoint(int splitId,
 FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>.ReaderAndOutput readerAndOutput)

Description copied from class: FlinkSourceReaderBase

Create FlinkSourceSplit for given splitId.

Specified by:

getReaderCheckpoint in class FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

createReader

protected Source.Reader<T> createReader(@Nonnull
 FlinkSourceSplit<T> sourceSplit)
                                 throws IOException

Description copied from class: FlinkSourceReaderBase

Create Source.Reader for given FlinkSourceSplit.

Specified by:

createReader in class FlinkSourceReaderBase<T,WindowedValue<ValueWithRecordId<T>>>

Throws:

IOException