org.apache.flink.streaming.connectors.kafka.internals

Class AbstractFetcher<T,KPH>

java.lang.Object

org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher<T,KPH>

Type Parameters:

T - The type of elements deserialized from Kafka's byte records, and emitted into the Flink data streams.

KPH - The type of topic/partition identifier used by Kafka in the specific version.

Direct Known Subclasses:

Kafka08Fetcher, Kafka09Fetcher

@Internal
public abstract class AbstractFetcher<T,KPH>
extends Object

Base class for all fetchers, which implement the connections to Kafka brokers and pull records from Kafka partitions.

This fetcher base class implements the logic around emitting records and tracking offsets, as well as around the optional timestamp assignment and watermark generation.

Field Summary

Fields

Modifier and Type	Field and Description
protected SourceFunction.SourceContext<T>	sourceContext The source context to emit records and watermarks to.
protected ClosableBlockingQueue<KafkaTopicPartitionState<KPH>>	unassignedPartitionsQueue Queue of partitions that are not yet assigned to any Kafka clients for consuming.

Constructor Summary

Constructors

Modifier

Constructor and Description

protected

AbstractFetcher(SourceFunction.SourceContext<T> sourceContext, Map<KafkaTopicPartition,Long> seedPartitionsWithInitialOffsets, SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic, SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated, ProcessingTimeService processingTimeProvider, long autoWatermarkInterval, ClassLoader userCodeClassLoader, MetricGroup consumerMetricGroup, boolean useMetrics)

Method Summary

Modifier and Type	Method and Description
void	addDiscoveredPartitions(List<KafkaTopicPartition> newPartitions) Adds a list of newly discovered partitions to the fetcher for consuming.
abstract void	cancel()
void	commitInternalOffsetsToKafka(Map<KafkaTopicPartition,Long> offsets, KafkaCommitCallback commitCallback) Commits the given partition offsets to the Kafka brokers (or to ZooKeeper for older Kafka versions).
protected abstract KPH	createKafkaPartitionHandle(KafkaTopicPartition partition) Creates the Kafka version specific representation of the given topic partition.
protected abstract void	doCommitInternalOffsetsToKafka(Map<KafkaTopicPartition,Long> offsets, KafkaCommitCallback commitCallback)
protected void	emitRecord(T record, KafkaTopicPartitionState<KPH> partitionState, long offset) Emits a record without attaching an existing timestamp to it.
protected void	emitRecordWithTimestamp(T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long timestamp) Emits a record attaching a timestamp to it.
abstract void	runFetchLoop()
HashMap<KafkaTopicPartition,Long>	snapshotCurrentState() Takes a snapshot of the partition offsets.
protected List<KafkaTopicPartitionState<KPH>>	subscribedPartitionStates() Gets all partitions (with partition state) that this fetcher is subscribed to.

Methods inherited from class java.lang.Object

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

Field Detail

sourceContext

protected final SourceFunction.SourceContext<T> sourceContext

The source context to emit records and watermarks to.

unassignedPartitionsQueue

protected final ClosableBlockingQueue<KafkaTopicPartitionState<KPH>> unassignedPartitionsQueue

Queue of partitions that are not yet assigned to any Kafka clients for consuming. Kafka version-specific implementations of runFetchLoop() should continuously poll this queue for unassigned partitions, and start consuming them accordingly.

All partitions added to this queue are guaranteed to have been added to subscribedPartitionStates already.

Constructor Detail

AbstractFetcher

protected AbstractFetcher(SourceFunction.SourceContext<T> sourceContext,
                          Map<KafkaTopicPartition,Long> seedPartitionsWithInitialOffsets,
                          SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
                          SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
                          ProcessingTimeService processingTimeProvider,
                          long autoWatermarkInterval,
                          ClassLoader userCodeClassLoader,
                          MetricGroup consumerMetricGroup,
                          boolean useMetrics)
                   throws Exception

Throws:

Exception

Method Detail

addDiscoveredPartitions

public void addDiscoveredPartitions(List<KafkaTopicPartition> newPartitions)
                             throws IOException,
                                    ClassNotFoundException

Adds a list of newly discovered partitions to the fetcher for consuming.

This method creates the partition state holder for each new partition, using KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET as the starting offset. It uses the earliest offset because there may be delay in discovering a partition after it was created and started receiving records.

After the state representation for a partition is created, it is added to the unassigned partitions queue to await to be consumed.

Parameters:

newPartitions - discovered partitions to add

Throws:

IOException

ClassNotFoundException

subscribedPartitionStates

protected final List<KafkaTopicPartitionState<KPH>> subscribedPartitionStates()

Gets all partitions (with partition state) that this fetcher is subscribed to.

Returns:

All subscribed partitions.

runFetchLoop

public abstract void runFetchLoop()
                           throws Exception

Throws:

Exception

cancel

public abstract void cancel()

commitInternalOffsetsToKafka

public final void commitInternalOffsetsToKafka(Map<KafkaTopicPartition,Long> offsets,
                                               @Nonnull
                                               KafkaCommitCallback commitCallback)
                                        throws Exception

Commits the given partition offsets to the Kafka brokers (or to ZooKeeper for older Kafka versions). This method is only ever called when the offset commit mode of the consumer is OffsetCommitMode.ON_CHECKPOINTS.

The given offsets are the internal checkpointed offsets, representing the last processed record of each partition. Version-specific implementations of this method need to hold the contract that the given offsets must be incremented by 1 before committing them, so that committed offsets to Kafka represent "the next record to process".

Parameters:

offsets - The offsets to commit to Kafka (implementations must increment offsets by 1 before committing).

commitCallback - The callback that the user should trigger when a commit request completes or fails.

Throws:

Exception - This method forwards exceptions.

doCommitInternalOffsetsToKafka

protected abstract void doCommitInternalOffsetsToKafka(Map<KafkaTopicPartition,Long> offsets,
                                                       @Nonnull
                                                       KafkaCommitCallback commitCallback)
                                                throws Exception

Throws:

Exception

createKafkaPartitionHandle

protected abstract KPH createKafkaPartitionHandle(KafkaTopicPartition partition)

Creates the Kafka version specific representation of the given topic partition.

Parameters:

partition - The Flink representation of the Kafka topic partition.

Returns:

The version-specific Kafka representation of the Kafka topic partition.

snapshotCurrentState

public HashMap<KafkaTopicPartition,Long> snapshotCurrentState()

Takes a snapshot of the partition offsets.

Important: This method must be called under the checkpoint lock.

Returns:

A map from partition to current offset.

emitRecord

protected void emitRecord(T record,
                          KafkaTopicPartitionState<KPH> partitionState,
                          long offset)
                   throws Exception

Emits a record without attaching an existing timestamp to it.

Implementation Note: This method is kept brief to be JIT inlining friendly. That makes the fast path efficient, the extended paths are called as separate methods.

Parameters:

record - The record to emit

partitionState - The state of the Kafka partition from which the record was fetched

offset - The offset of the record

Throws:

Exception

emitRecordWithTimestamp

protected void emitRecordWithTimestamp(T record,
                                       KafkaTopicPartitionState<KPH> partitionState,
                                       long offset,
                                       long timestamp)
                                throws Exception

Emits a record attaching a timestamp to it.

Implementation Note: This method is kept brief to be JIT inlining friendly. That makes the fast path efficient, the extended paths are called as separate methods.

Parameters:

record - The record to emit

partitionState - The state of the Kafka partition from which the record was fetched

offset - The offset of the record

Throws:

Exception