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

Class AbstractPartitionDiscoverer

java.lang.Object

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

Direct Known Subclasses:

KafkaPartitionDiscoverer

@Internal
public abstract class AbstractPartitionDiscoverer
extends Object

Base class for all partition discoverers.

This partition discoverer base class implements the logic around bookkeeping discovered partitions, and using the information to determine whether or not there are new partitions that the consumer subtask should subscribe to.

Subclass implementations should simply implement the logic of using the version-specific Kafka clients to fetch topic and partition metadata.

Since Kafka clients are generally not thread-safe, partition discoverers should not be concurrently accessed. The only exception for this would be the wakeup() call, which allows the discoverer to be interrupted during a discoverPartitions() call.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	AbstractPartitionDiscoverer.ClosedException Thrown if this discoverer was used to discover partitions after it was closed.
static class	AbstractPartitionDiscoverer.WakeupException Signaling exception to indicate that an actual Kafka call was interrupted.

Constructor Summary

Constructors

Constructor and Description
AbstractPartitionDiscoverer(KafkaTopicsDescriptor topicsDescriptor, int indexOfThisSubtask, int numParallelSubtasks)

Method Summary

Modifier and Type	Method and Description
void	close() Closes the partition discoverer, cleaning up all Kafka connections.
protected abstract void	closeConnections() Close all established connections.
List<KafkaTopicPartition>	discoverPartitions() Execute a partition discovery attempt for this subtask.
protected abstract List<KafkaTopicPartition>	getAllPartitionsForTopics(List<String> topics) Fetch the list of all partitions for a specific topics list from Kafka.
protected abstract List<String>	getAllTopics() Fetch the list of all topics from Kafka.
protected abstract void	initializeConnections() Establish the required connections in order to fetch topics and partitions metadata.
void	open() Opens the partition discoverer, initializing all required Kafka connections.
boolean	setAndCheckDiscoveredPartition(KafkaTopicPartition partition) Sets a partition as discovered.
void	wakeup() Interrupt an in-progress discovery attempt by throwing a AbstractPartitionDiscoverer.WakeupException.
protected abstract void	wakeupConnections() Attempt to eagerly wakeup from blocking calls to Kafka in getAllTopics() and getAllPartitionsForTopics(List).

Methods inherited from class java.lang.Object

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

Constructor Detail

AbstractPartitionDiscoverer

public AbstractPartitionDiscoverer(KafkaTopicsDescriptor topicsDescriptor,
                                   int indexOfThisSubtask,
                                   int numParallelSubtasks)

Method Detail

open

public void open()
          throws Exception

Opens the partition discoverer, initializing all required Kafka connections.

NOTE: thread-safety is not guaranteed.

Throws:

Exception

close

public void close()
           throws Exception

Closes the partition discoverer, cleaning up all Kafka connections.

NOTE: thread-safety is not guaranteed.

Throws:

Exception

wakeup

public void wakeup()

Interrupt an in-progress discovery attempt by throwing a AbstractPartitionDiscoverer.WakeupException. If no attempt is in progress, the immediate next attempt will throw a AbstractPartitionDiscoverer.WakeupException.

This method can be called concurrently from a different thread.

discoverPartitions

public List<KafkaTopicPartition> discoverPartitions()
                                             throws AbstractPartitionDiscoverer.WakeupException,
                                                    AbstractPartitionDiscoverer.ClosedException

Execute a partition discovery attempt for this subtask. This method lets the partition discoverer update what partitions it has discovered so far.

Returns:

List of discovered new partitions that this subtask should subscribe to.

Throws:

AbstractPartitionDiscoverer.WakeupException

AbstractPartitionDiscoverer.ClosedException

setAndCheckDiscoveredPartition

public boolean setAndCheckDiscoveredPartition(KafkaTopicPartition partition)

Sets a partition as discovered. Partitions are considered as new if its partition id is larger than all partition ids previously seen for the topic it belongs to. Therefore, for a set of discovered partitions, the order that this method is invoked with each partition is important.

If the partition is indeed newly discovered, this method also returns whether the new partition should be subscribed by this subtask.

Parameters:

partition - the partition to set and check

Returns:

true, if the partition wasn't seen before and should be subscribed by this subtask; false otherwise

initializeConnections

protected abstract void initializeConnections()
                                       throws Exception

Establish the required connections in order to fetch topics and partitions metadata.

Throws:

Exception

wakeupConnections

protected abstract void wakeupConnections()

Attempt to eagerly wakeup from blocking calls to Kafka in getAllTopics() and getAllPartitionsForTopics(List).

If the invocation indeed results in interrupting an actual blocking Kafka call, the implementations of getAllTopics() and getAllPartitionsForTopics(List) are responsible of throwing a AbstractPartitionDiscoverer.WakeupException.

closeConnections

protected abstract void closeConnections()
                                  throws Exception

Close all established connections.

Throws:

Exception

getAllTopics

protected abstract List<String> getAllTopics()
                                      throws AbstractPartitionDiscoverer.WakeupException

Fetch the list of all topics from Kafka.

Throws:

AbstractPartitionDiscoverer.WakeupException

getAllPartitionsForTopics

protected abstract List<KafkaTopicPartition> getAllPartitionsForTopics(List<String> topics)
                                                                throws AbstractPartitionDiscoverer.WakeupException

Fetch the list of all partitions for a specific topics list from Kafka.

Throws:

AbstractPartitionDiscoverer.WakeupException