org.apache.flink.streaming.connectors.kafka

Class FlinkKafkaConsumer010<T>

java.lang.Object

org.apache.flink.api.common.functions.AbstractRichFunction

org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction<T>

org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase<T>

org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09<T>

org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010<T>

All Implemented Interfaces:

Serializable, Function, RichFunction, ResultTypeQueryable<T>, CheckpointListener, CheckpointedFunction, ParallelSourceFunction<T>, SourceFunction<T>

Direct Known Subclasses:

FlinkKafkaConsumer011

@PublicEvolving
public class FlinkKafkaConsumer010<T>
extends FlinkKafkaConsumer09<T>

The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from Apache Kafka 0.10.x. The consumer can run in multiple parallel instances, each of which will pull data from one or more Kafka partitions.

The Flink Kafka Consumer participates in checkpointing and guarantees that no data is lost during a failure, and that the computation processes elements "exactly once". (Note: These guarantees naturally assume that Kafka itself does not loose any data.)

Please note that Flink snapshots the offsets internally as part of its distributed checkpoints. The offsets committed to Kafka / ZooKeeper are only to bring the outside view of progress in sync with Flink's view of the progress. That way, monitoring and other jobs can get a view of how far the Flink Kafka consumer has consumed a topic.

Please refer to Kafka's documentation for the available configuration properties: http://kafka.apache.org/documentation.html#newconsumerconfigs

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.streaming.api.functions.source.SourceFunction

SourceFunction.SourceContext<T>

Field Summary

Fields inherited from class org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09

DEFAULT_POLL_TIMEOUT, KEY_POLL_TIMEOUT, pollTimeout, properties

Fields inherited from class org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase

deserializer, KEY_DISABLE_METRICS, KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, MAX_NUM_PENDING_CHECKPOINTS, PARTITION_DISCOVERY_DISABLED

Constructor Summary

Constructors

Constructor and Description
FlinkKafkaConsumer010(List<String> topics, DeserializationSchema<T> deserializer, Properties props) Creates a new Kafka streaming source consumer for Kafka 0.10.x
FlinkKafkaConsumer010(List<String> topics, KafkaDeserializationSchema<T> deserializer, Properties props) Creates a new Kafka streaming source consumer for Kafka 0.10.x
FlinkKafkaConsumer010(Pattern subscriptionPattern, DeserializationSchema<T> valueDeserializer, Properties props) Creates a new Kafka streaming source consumer for Kafka 0.10.x.
FlinkKafkaConsumer010(Pattern subscriptionPattern, KafkaDeserializationSchema<T> deserializer, Properties props) Creates a new Kafka streaming source consumer for Kafka 0.10.x.
FlinkKafkaConsumer010(String topic, DeserializationSchema<T> valueDeserializer, Properties props) Creates a new Kafka streaming source consumer for Kafka 0.10.x.
FlinkKafkaConsumer010(String topic, KafkaDeserializationSchema<T> deserializer, Properties props) Creates a new Kafka streaming source consumer for Kafka 0.10.x

Method Summary

Modifier and Type	Method and Description
protected AbstractFetcher<T,?>	createFetcher(SourceFunction.SourceContext<T> sourceContext, Map<KafkaTopicPartition,Long> assignedPartitionsWithInitialOffsets, SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic, SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated, StreamingRuntimeContext runtimeContext, OffsetCommitMode offsetCommitMode, MetricGroup consumerMetricGroup, boolean useMetrics) Creates the fetcher that connect to the Kafka brokers, pulls data, deserialized the data, and emits it into the data streams.
protected AbstractPartitionDiscoverer	createPartitionDiscoverer(KafkaTopicsDescriptor topicsDescriptor, int indexOfThisSubtask, int numParallelSubtasks) Creates the partition discoverer that is used to find new partitions for this subtask.
protected Map<KafkaTopicPartition,Long>	fetchOffsetsWithTimestamp(Collection<KafkaTopicPartition> partitions, long timestamp)
FlinkKafkaConsumerBase<T>	setStartFromTimestamp(long startupOffsetsTimestamp) Specifies the consumer to start reading partitions from a specified timestamp.

Methods inherited from class org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09

getIsAutoCommitEnabled, getRateLimiter, setRateLimiter

Methods inherited from class org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase

assignTimestampsAndWatermarks, assignTimestampsAndWatermarks, cancel, close, disableFilterRestoredPartitionsWithSubscribedTopics, getProducedType, initializeState, notifyCheckpointComplete, open, run, setCommitOffsetsOnCheckpoints, setStartFromEarliest, setStartFromGroupOffsets, setStartFromLatest, setStartFromSpecificOffsets, snapshotState

Methods inherited from class org.apache.flink.api.common.functions.AbstractRichFunction

getIterationRuntimeContext, getRuntimeContext, setRuntimeContext

Methods inherited from class java.lang.Object

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

Constructor Detail

FlinkKafkaConsumer010

public FlinkKafkaConsumer010(String topic,
                             DeserializationSchema<T> valueDeserializer,
                             Properties props)

Creates a new Kafka streaming source consumer for Kafka 0.10.x.

Parameters:

topic - The name of the topic that should be consumed.

valueDeserializer - The de-/serializer used to convert between Kafka's byte messages and Flink's objects.

props - The properties used to configure the Kafka consumer client, and the ZooKeeper client.

FlinkKafkaConsumer010

public FlinkKafkaConsumer010(String topic,
                             KafkaDeserializationSchema<T> deserializer,
                             Properties props)

Creates a new Kafka streaming source consumer for Kafka 0.10.x

This constructor allows passing a KafkaDeserializationSchema for reading key/value pairs, offsets, and topic names from Kafka.

Parameters:

topic - The name of the topic that should be consumed.

deserializer - The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects.

props - The properties used to configure the Kafka consumer client, and the ZooKeeper client.

FlinkKafkaConsumer010

public FlinkKafkaConsumer010(List<String> topics,
                             DeserializationSchema<T> deserializer,
                             Properties props)

Creates a new Kafka streaming source consumer for Kafka 0.10.x

This constructor allows passing multiple topics to the consumer.

Parameters:

topics - The Kafka topics to read from.

deserializer - The de-/serializer used to convert between Kafka's byte messages and Flink's objects.

props - The properties that are used to configure both the fetcher and the offset handler.

FlinkKafkaConsumer010

public FlinkKafkaConsumer010(List<String> topics,
                             KafkaDeserializationSchema<T> deserializer,
                             Properties props)

Creates a new Kafka streaming source consumer for Kafka 0.10.x

This constructor allows passing multiple topics and a key/value deserialization schema.

Parameters:

topics - The Kafka topics to read from.

deserializer - The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects.

props - The properties that are used to configure both the fetcher and the offset handler.

FlinkKafkaConsumer010

@PublicEvolving
public FlinkKafkaConsumer010(Pattern subscriptionPattern,
                                             DeserializationSchema<T> valueDeserializer,
                                             Properties props)

Creates a new Kafka streaming source consumer for Kafka 0.10.x. Use this constructor to subscribe to multiple topics based on a regular expression pattern.

If partition discovery is enabled (by setting a non-negative value for FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS in the properties), topics with names matching the pattern will also be subscribed to as they are created on the fly.

Parameters:

subscriptionPattern - The regular expression for a pattern of topic names to subscribe to.

valueDeserializer - The de-/serializer used to convert between Kafka's byte messages and Flink's objects.

props - The properties used to configure the Kafka consumer client, and the ZooKeeper client.

FlinkKafkaConsumer010

@PublicEvolving
public FlinkKafkaConsumer010(Pattern subscriptionPattern,
                                             KafkaDeserializationSchema<T> deserializer,
                                             Properties props)

Creates a new Kafka streaming source consumer for Kafka 0.10.x. Use this constructor to subscribe to multiple topics based on a regular expression pattern.

If partition discovery is enabled (by setting a non-negative value for FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS in the properties), topics with names matching the pattern will also be subscribed to as they are created on the fly.

This constructor allows passing a KafkaDeserializationSchema for reading key/value pairs, offsets, and topic names from Kafka.

Parameters:

subscriptionPattern - The regular expression for a pattern of topic names to subscribe to.

deserializer - The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects.

props - The properties used to configure the Kafka consumer client, and the ZooKeeper client.

Method Detail

createFetcher

protected AbstractFetcher<T,?> createFetcher(SourceFunction.SourceContext<T> sourceContext,
                                             Map<KafkaTopicPartition,Long> assignedPartitionsWithInitialOffsets,
                                             SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic,
                                             SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated,
                                             StreamingRuntimeContext runtimeContext,
                                             OffsetCommitMode offsetCommitMode,
                                             MetricGroup consumerMetricGroup,
                                             boolean useMetrics)
                                      throws Exception

Description copied from class: FlinkKafkaConsumerBase

Creates the fetcher that connect to the Kafka brokers, pulls data, deserialized the data, and emits it into the data streams.

Overrides:

createFetcher in class FlinkKafkaConsumer09<T>

Parameters:

sourceContext - The source context to emit data to.

assignedPartitionsWithInitialOffsets - The set of partitions that this subtask should handle, with their start offsets.

watermarksPeriodic - Optional, a serialized timestamp extractor / periodic watermark generator.

watermarksPunctuated - Optional, a serialized timestamp extractor / punctuated watermark generator.

runtimeContext - The task's runtime context.

Returns:

The instantiated fetcher

Throws:

Exception - The method should forward exceptions

createPartitionDiscoverer

protected AbstractPartitionDiscoverer createPartitionDiscoverer(KafkaTopicsDescriptor topicsDescriptor,
                                                                int indexOfThisSubtask,
                                                                int numParallelSubtasks)

Description copied from class: FlinkKafkaConsumerBase

Creates the partition discoverer that is used to find new partitions for this subtask.

Overrides:

createPartitionDiscoverer in class FlinkKafkaConsumer09<T>

Parameters:

topicsDescriptor - Descriptor that describes whether we are discovering partitions for fixed topics or a topic pattern.

indexOfThisSubtask - The index of this consumer subtask.

numParallelSubtasks - The total number of parallel consumer subtasks.

Returns:

The instantiated partition discoverer

setStartFromTimestamp

public FlinkKafkaConsumerBase<T> setStartFromTimestamp(long startupOffsetsTimestamp)

Description copied from class: FlinkKafkaConsumerBase

Specifies the consumer to start reading partitions from a specified timestamp. The specified timestamp must be before the current timestamp. This lets the consumer ignore any committed group offsets in Zookeeper / Kafka brokers.

The consumer will look up the earliest offset whose timestamp is greater than or equal to the specific timestamp from Kafka. If there's no such offset, the consumer will use the latest offset to read data from kafka.

This method does not affect where partitions are read from when the consumer is restored from a checkpoint or savepoint. When the consumer is restored from a checkpoint or savepoint, only the offsets in the restored state will be used.

Overrides:

setStartFromTimestamp in class FlinkKafkaConsumerBase<T>

Parameters:

startupOffsetsTimestamp - timestamp for the startup offsets, as milliseconds from epoch.

Returns:

The consumer object, to allow function chaining.

fetchOffsetsWithTimestamp

protected Map<KafkaTopicPartition,Long> fetchOffsetsWithTimestamp(Collection<KafkaTopicPartition> partitions,
                                                                  long timestamp)

Overrides:

fetchOffsetsWithTimestamp in class FlinkKafkaConsumer09<T>