org.apache.flink.connector.kafka.source.reader

Class KafkaSourceReader<T>

java.lang.Object

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

org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase<Tuple3<T,Long,Long>,T,KafkaPartitionSplit,KafkaPartitionSplitState>

org.apache.flink.connector.kafka.source.reader.KafkaSourceReader<T>

All Implemented Interfaces:

AutoCloseable, CheckpointListener, SourceReader<T,KafkaPartitionSplit>

public class KafkaSourceReader<T>
extends SingleThreadMultiplexSourceReaderBase<Tuple3<T,Long,Long>,T,KafkaPartitionSplit,KafkaPartitionSplitState>

The source reader for Kafka partitions.

Field Summary

Fields inherited from class org.apache.flink.connector.base.source.reader.SourceReaderBase

config, context, options, recordEmitter, splitFetcherManager

Constructor Summary

Constructors

Constructor and Description
KafkaSourceReader(FutureCompletingBlockingQueue<RecordsWithSplitIds<Tuple3<T,Long,Long>>> elementsQueue, java.util.function.Supplier<KafkaPartitionSplitReader<T>> splitReaderSupplier, RecordEmitter<Tuple3<T,Long,Long>,T,KafkaPartitionSplitState> recordEmitter, Configuration config, SourceReaderContext context)

Method Summary

Modifier and Type	Method and Description
protected KafkaPartitionSplitState	initializedState(KafkaPartitionSplit split) When new splits are added to the reader.
void	notifyCheckpointComplete(long checkpointId) We have an empty default implementation here because most source readers do not have to implement the method.
protected void	onSplitFinished(Map<String,KafkaPartitionSplitState> finishedSplitIds) Handles the finished splits to clean the state if needed.
List<KafkaPartitionSplit>	snapshotState(long checkpointId) Checkpoint on the state of the source.
protected KafkaPartitionSplit	toSplitType(String splitId, KafkaPartitionSplitState splitState) Convert a mutable SplitStateT to immutable SplitT.

Methods inherited from class org.apache.flink.connector.base.source.reader.SourceReaderBase

addSplits, close, getNumberOfCurrentlyAssignedSplits, handleSourceEvents, isAvailable, notifyNoMoreSplits, pollNext, start

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

Constructor Detail

KafkaSourceReader

public KafkaSourceReader(FutureCompletingBlockingQueue<RecordsWithSplitIds<Tuple3<T,Long,Long>>> elementsQueue,
                         java.util.function.Supplier<KafkaPartitionSplitReader<T>> splitReaderSupplier,
                         RecordEmitter<Tuple3<T,Long,Long>,T,KafkaPartitionSplitState> recordEmitter,
                         Configuration config,
                         SourceReaderContext context)

Method Detail

onSplitFinished

protected void onSplitFinished(Map<String,KafkaPartitionSplitState> finishedSplitIds)

Description copied from class: SourceReaderBase

Handles the finished splits to clean the state if needed.

Specified by:

onSplitFinished in class SourceReaderBase<Tuple3<T,Long,Long>,T,KafkaPartitionSplit,KafkaPartitionSplitState>

snapshotState

public List<KafkaPartitionSplit> snapshotState(long checkpointId)

Description copied from interface: SourceReader

Checkpoint on the state of the source.

Specified by:

snapshotState in interface SourceReader<T,KafkaPartitionSplit>

Overrides:

snapshotState in class SourceReaderBase<Tuple3<T,Long,Long>,T,KafkaPartitionSplit,KafkaPartitionSplitState>

Returns:

the state of the source.

notifyCheckpointComplete

public void notifyCheckpointComplete(long checkpointId)
                              throws Exception

Description copied from interface: SourceReader

We have an empty default implementation here because most source readers do not have to implement the method.

Parameters:

checkpointId - The ID of the checkpoint that has been completed.

Throws:

Exception - This method can propagate exceptions, which leads to a failure/recovery for the task. Not that this will NOT lead to the checkpoint being revoked.

See Also:

CheckpointListener.notifyCheckpointComplete(long)

initializedState

protected KafkaPartitionSplitState initializedState(KafkaPartitionSplit split)

Description copied from class: SourceReaderBase

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

Specified by:

initializedState in class SourceReaderBase<Tuple3<T,Long,Long>,T,KafkaPartitionSplit,KafkaPartitionSplitState>

Parameters:

split - a newly added split.

toSplitType

protected KafkaPartitionSplit toSplitType(String splitId,
                                          KafkaPartitionSplitState splitState)

Description copied from class: SourceReaderBase

Convert a mutable SplitStateT to immutable SplitT.

Specified by:

toSplitType in class SourceReaderBase<Tuple3<T,Long,Long>,T,KafkaPartitionSplit,KafkaPartitionSplitState>

splitState - splitState.

Returns:

an immutable Split state.