org.apache.kafka.clients.consumer

Class KafkaConsumer<K,V>

java.lang.Object

org.apache.kafka.clients.consumer.KafkaConsumer<K,V>

All Implemented Interfaces:

java.io.Closeable, java.lang.AutoCloseable, Consumer<K,V>

@InterfaceStability.Unstable
public class KafkaConsumer<K,V>
extends java.lang.Object
implements Consumer<K,V>

A Kafka client that consumes records from a Kafka cluster.

It will transparently handle the failure of servers in the Kafka cluster, and transparently adapt as partitions of data it fetches migrate within the cluster. This client also interacts with the server to allow groups of consumers to load balance consumption using consumer groups (as described below).

The consumer maintains TCP connections to the necessary brokers to fetch data. Failure to close the consumer after use will leak these connections. The consumer is not thread-safe. See Multi-threaded Processing for more details.

Offsets and Consumer Position

Kafka maintains a numerical offset for each record in a partition. This offset acts as a kind of unique identifier of a record within that partition, and also denotes the position of the consumer in the partition. That is, a consumer which has position 5 has consumed records with offsets 0 through 4 and will next receive record with offset 5. There are actually two notions of position relevant to the user of the consumer.

The position of the consumer gives the offset of the next record that will be given out. It will be one larger than the highest offset the consumer has seen in that partition. It automatically advances every time the consumer receives data calls poll(long) and receives messages.

The committed position is the last offset that has been saved securely. Should the process fail and restart, this is the offset that it will recover to. The consumer can either automatically commit offsets periodically; or it can choose to control this committed position manually by calling commitSync, which will block until the offsets have been successfully committed or fatal error has happened during the commit process, or commitAsync which is non-blocking and will trigger OffsetCommitCallback upon either successfully committed or fatally failed.

This distinction gives the consumer control over when a record is considered consumed. It is discussed in further detail below.

Consumer Groups and Topic Subscriptions

Kafka uses the concept of consumer groups to allow a pool of processes to divide up the work of consuming and processing records. These processes can either be running on the same machine or, as is more likely, they can be distributed over many machines to provide additional scalability and fault tolerance for processing.

Each Kafka consumer is able to configure a consumer group that it belongs to, and can dynamically set the list of topics it wants to subscribe to through subscribe(List, ConsumerRebalanceListener), or subscribe to all topics matching certain pattern through subscribe(Pattern, ConsumerRebalanceListener). Kafka will deliver each message in the subscribed topics to one process in each consumer group. This is achieved by balancing the partitions in the topic over the consumer processes in each group. So if there is a topic with four partitions, and a consumer group with two processes, each process would consume from two partitions. This group membership is maintained dynamically: if a process fails the partitions assigned to it will be reassigned to other processes in the same group, and if a new process joins the group, partitions will be moved from existing consumers to this new process.

So if two processes subscribe to a topic both specifying different groups they will each get all the records in that topic; if they both specify the same group they will each get about half the records.

Conceptually you can think of a consumer group as being a single logical subscriber that happens to be made up of multiple processes. As a multi-subscriber system, Kafka naturally supports having any number of consumer groups for a given topic without duplicating data (additional consumers are actually quite cheap).

This is a slight generalization of the functionality that is common in messaging systems. To get semantics similar to a queue in a traditional messaging system all processes would be part of a single consumer group and hence record delivery would be balanced over the group like with a queue. Unlike a traditional messaging system, though, you can have multiple such groups. To get semantics similar to pub-sub in a traditional messaging system each process would have its own consumer group, so each process would subscribe to all the records published to the topic.

In addition, when group reassignment happens automatically, consumers can be notified through ConsumerRebalanceListener, which allows them to finish necessary application-level logic such as state cleanup, manual offset commits (note that offsets are always committed for a given consumer group), etc. See Storing Offsets Outside Kafka for more details

It is also possible for the consumer to manually specify the partitions that are assigned to it through assign(List), which disables this dynamic partition assignment.

Usage Examples

The consumer APIs offer flexibility to cover a variety of consumption use cases. Here are some examples to demonstrate how to use them.

Automatic Offset Committing

This example demonstrates a simple usage of Kafka's consumer api that relying on automatic offset committing.

     Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "true");
     props.put("auto.commit.interval.ms", "1000");
     props.put("session.timeout.ms", "30000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("foo", "bar"));
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
             System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
     }
 

Setting enable.auto.commit means that offsets are committed automatically with a frequency controlled by the config auto.commit.interval.ms.

The connection to the cluster is bootstrapped by specifying a list of one or more brokers to contact using the configuration bootstrap.servers. This list is just used to discover the rest of the brokers in the cluster and need not be an exhaustive list of servers in the cluster (though you may want to specify more than one in case there are servers down when the client is connecting).

In this example the client is subscribing to the topics foo and bar as part of a group of consumers called test as described above.

The broker will automatically detect failed processes in the test group by using a heartbeat mechanism. The consumer will automatically ping the cluster periodically, which lets the cluster know that it is alive. As long as the consumer is able to do this it is considered alive and retains the right to consume from the partitions assigned to it. If it stops heartbeating for a period of time longer than session.timeout.ms then it will be considered dead and its partitions will be assigned to another process.

The deserializer settings specify how to turn bytes into objects. For example, by specifying string deserializers, we are saying that our record's key and value will just be simple strings.

Manual Offset Control

Instead of relying on the consumer to periodically commit consumed offsets, users can also control when messages should be considered as consumed and hence commit their offsets. This is useful when the consumption of the messages are coupled with some processing logic and hence a message should not be considered as consumed until it is completed processing. In this example we will consume a batch of records and batch them up in memory, when we have sufficient records batched we will insert them into a database. If we allowed offsets to auto commit as in the previous example messages would be considered consumed after they were given out by the consumer, and it would be possible that our process could fail after we have read messages into our in-memory buffer but before they had been inserted into the database. To avoid this we will manually commit the offsets only once the corresponding messages have been inserted into the database. This gives us exact control of when a message is considered consumed. This raises the opposite possibility: the process could fail in the interval after the insert into the database but before the commit (even though this would likely just be a few milliseconds, it is a possibility). In this case the process that took over consumption would consume from last committed offset and would repeat the insert of the last batch of data. Used in this way Kafka provides what is often called "at-least once delivery" guarantees, as each message will likely be delivered one time but in failure cases could be duplicated.

     Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "false");
     props.put("auto.commit.interval.ms", "1000");
     props.put("session.timeout.ms", "30000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("foo", "bar"));
     final int minBatchSize = 200;
     List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records) {
             buffer.add(record);
         }
         if (buffer.size() >= minBatchSize) {
             insertIntoDb(buffer);
             consumer.commitSync();
             buffer.clear();
         }
     }
 

The above example uses commitSync to mark all received messages as committed. In some cases you may wish to have even finer control over which messages have been committed by specifying an offset explicitly. In the example below we commit offset after we finish handling the messages in each partition.

     try {
         while(running) {
             ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
             for (TopicPartition partition : records.partitions()) {
                 List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                 for (ConsumerRecord<String, String> record : partitionRecords) {
                     System.out.println(record.offset() + ": " + record.value());
                 }
                 long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                 consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
             }
         }
     } finally {
       consumer.close();
     }
 

Note: The committed offset should always be the offset of the next message that your application will read. Thus, when calling commitSync(offsets) you should add one to the offset of the last message processed.

Subscribing To Specific Partitions

In the previous examples we subscribed to the topics we were interested in and let Kafka give our particular process a fair share of the partitions for those topics. This provides a simple load balancing mechanism so multiple instances of our program can divided up the work of processing records.

In this mode the consumer will just get the partitions it subscribes to and if the consumer instance fails no attempt will be made to rebalance partitions to other instances.

There are several cases where this makes sense:

• The first case is if the process is maintaining some kind of local state associated with that partition (like a local on-disk key-value store) and hence it should only get records for the partition it is maintaining on disk.
• Another case is if the process itself is highly available and will be restarted if it fails (perhaps using a cluster management framework like YARN, Mesos, or AWS facilities, or as part of a stream processing framework). In this case there is no need for Kafka to detect the failure and reassign the partition, rather the consuming process will be restarted on another machine.

This mode is easy to specify, rather than subscribing to the topic, the consumer just subscribes to particular partitions:

     String topic = "foo";
     TopicPartition partition0 = new TopicPartition(topic, 0);
     TopicPartition partition1 = new TopicPartition(topic, 1);
     consumer.assign(Arrays.asList(partition0, partition1));
 

The group that the consumer specifies is still used for committing offsets, but now the set of partitions will only be changed if the consumer specifies new partitions, and no attempt at failure detection will be made.

It isn't possible to mix both subscription to specific partitions (with no load balancing) and to topics (with load balancing) using the same consumer instance.

Storing Offsets Outside Kafka

The consumer application need not use Kafka's built-in offset storage, it can store offsets in a store of its own choosing. The primary use case for this is allowing the application to store both the offset and the results of the consumption in the same system in a way that both the results and offsets are stored atomically. This is not always possible, but when it is it will make the consumption fully atomic and give "exactly once" semantics that are stronger than the default "at-least once" semantics you get with Kafka's offset commit functionality.

Here are a couple of examples of this type of usage:

• If the results of the consumption are being stored in a relational database, storing the offset in the database as well can allow committing both the results and offset in a single transaction. Thus either the transaction will succeed and the offset will be updated based on what was consumed or the result will not be stored and the offset won't be updated.
• If the results are being stored in a local store it may be possible to store the offset there as well. For example a search index could be built by subscribing to a particular partition and storing both the offset and the indexed data together. If this is done in a way that is atomic, it is often possible to have it be the case that even if a crash occurs that causes unsync'd data to be lost, whatever is left has the corresponding offset stored as well. This means that in this case the indexing process that comes back having lost recent updates just resumes indexing from what it has ensuring that no updates are lost.

Each record comes with its own offset, so to manage your own offset you just need to do the following:

• Configure enable.auto.commit=false
• Use the offset provided with each ConsumerRecord to save your position.
• On restart restore the position of the consumer using seek(TopicPartition, long).

This type of usage is simplest when the partition assignment is also done manually (this would be likely in the search index use case described above). If the partition assignment is done automatically special care is needed to handle the case where partition assignments change. This can be done by providing a ConsumerRebalanceListener instance in the call to subscribe(List, ConsumerRebalanceListener) and subscribe(Pattern, ConsumerRebalanceListener). For example, when partitions are taken from a consumer the consumer will want to commit its offset for those partitions by implementing ConsumerRebalanceListener.onPartitionsRevoked(Collection). When partitions are assigned to a consumer, the consumer will want to look up the offset for those new partitions and correctly initialize the consumer to that position by implementing ConsumerRebalanceListener.onPartitionsAssigned(Collection).

Another common use for ConsumerRebalanceListener is to flush any caches the application maintains for partitions that are moved elsewhere.

Controlling The Consumer's Position

In most use cases the consumer will simply consume records from beginning to end, periodically committing its position (either automatically or manually). However Kafka allows the consumer to manually control its position, moving forward or backwards in a partition at will. This means a consumer can re-consume older records, or skip to the most recent records without actually consuming the intermediate records.

There are several instances where manually controlling the consumer's position can be useful.

One case is for time-sensitive record processing it may make sense for a consumer that falls far enough behind to not attempt to catch up processing all records, but rather just skip to the most recent records.

Another use case is for a system that maintains local state as described in the previous section. In such a system the consumer will want to initialize its position on start-up to whatever is contained in the local store. Likewise if the local state is destroyed (say because the disk is lost) the state may be recreated on a new machine by re-consuming all the data and recreating the state (assuming that Kafka is retaining sufficient history).

Kafka allows specifying the position using seek(TopicPartition, long) to specify the new position. Special methods for seeking to the earliest and latest offset the server maintains are also available ( seekToBeginning(TopicPartition...) and seekToEnd(TopicPartition...) respectively).

Consumption Flow Control

If a consumer is assigned multiple partitions to fetch data from, it will try to consume from all of them at the same time, effectively giving these partitions the same priority for consumption. However in some cases consumers may want to first focus on fetching from some subset of the assigned partitions at full speed, and only start fetching other partitions when these partitions have few or no data to consume.

One of such cases is stream processing, where processor fetches from two topics and performs the join on these two streams. When one of the topic is long lagging behind the other, the processor would like to pause fetching from the ahead topic in order to get the lagging stream to catch up. Another example is bootstraping upon consumer starting up where there are a lot of history data to catch up, the applciations usually wants to get the latest data on some of the topics before consider fetching other topics.

Kafka supports dynamic controlling of consumption flows by using pause(TopicPartition...) and resume(TopicPartition...) to pause the consumption on the specified assigned partitions and resume the consumption on the specified paused partitions respectively in the future poll(long) calls.

Multi-threaded Processing

The Kafka consumer is NOT thread-safe. All network I/O happens in the thread of the application making the call. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized. Un-synchronized access will result in ConcurrentModificationException.

The only exception to this rule is wakeup(), which can safely be used from an external thread to interrupt an active operation. In this case, a WakeupException will be thrown from the thread blocking on the operation. This can be used to shutdown the consumer from another thread. The following snippet shows the typical pattern:

 public class KafkaConsumerRunner implements Runnable {
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final KafkaConsumer consumer;

     public void run() {
         try {
             consumer.subscribe(Arrays.asList("topic"));
             while (!closed.get()) {
                 ConsumerRecords records = consumer.poll(10000);
                 // Handle new records
             }
         } catch (WakeupException e) {
             // Ignore exception if closing
             if (!closed.get()) throw e;
         } finally {
             consumer.close();
         }
     }

     // Shutdown hook which can be called from a separate thread
     public void shutdown() {
         closed.set(true);
         consumer.wakeup();
     }
 }
 

Then in a separate thread, the consumer can be shutdown by setting the closed flag and waking up the consumer.

     closed.set(true);
     consumer.wakeup();
 

We have intentionally avoided implementing a particular threading model for processing. This leaves several options for implementing multi-threaded processing of records.

1. One Consumer Per Thread

A simple option is to give each thread its own consumer instance. Here are the pros and cons of this approach:

• PRO: It is the easiest to implement
• PRO: It is often the fastest as no inter-thread co-ordination is needed
• PRO: It makes in-order processing on a per-partition basis very easy to implement (each thread just processes messages in the order it receives them).
• CON: More consumers means more TCP connections to the cluster (one per thread). In general Kafka handles connections very efficiently so this is generally a small cost.
• CON: Multiple consumers means more requests being sent to the server and slightly less batching of data which can cause some drop in I/O throughput.
• CON: The number of total threads across all processes will be limited by the total number of partitions.

2. Decouple Consumption and Processing

Another alternative is to have one or more consumer threads that do all data consumption and hands off ConsumerRecords instances to a blocking queue consumed by a pool of processor threads that actually handle the record processing. This option likewise has pros and cons:

• PRO: This option allows independently scaling the number of consumers and processors. This makes it possible to have a single consumer that feeds many processor threads, avoiding any limitation on partitions.
• CON: Guaranteeing order across the processors requires particular care as the threads will execute independently an earlier chunk of data may actually be processed after a later chunk of data just due to the luck of thread execution timing. For processing that has no ordering requirements this is not a problem.
• CON: Manually committing the position becomes harder as it requires that all threads co-ordinate to ensure that processing is complete for that partition.

There are many possible variations on this approach. For example each processor thread can have its own queue, and the consumer threads can hash into these queues using the TopicPartition to ensure in-order consumption and simplify commit.

Constructor Summary

Constructors

Constructor and Description
KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs) A consumer is instantiated by providing a set of key-value pairs as configuration.
KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) A consumer is instantiated by providing a set of key-value pairs as configuration, a ConsumerRebalanceListener implementation, a key and a value Deserializer.
KafkaConsumer(java.util.Properties properties) A consumer is instantiated by providing a Properties object as configuration.
KafkaConsumer(java.util.Properties properties, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) A consumer is instantiated by providing a Properties object as configuration and a ConsumerRebalanceListener implementation, a key and a value Deserializer.

Method Summary

Methods

Modifier and Type	Method and Description
void	assign(java.util.List<TopicPartition> partitions) Manually assign a list of partition to this consumer.
java.util.Set<TopicPartition>	assignment() Get the set of partitions currently assigned to this consumer.
void	close() Close the consumer, waiting indefinitely for any needed cleanup.
void	commitAsync() Commit offsets returned on the last poll() for all the subscribed list of topics and partition.
void	commitAsync(java.util.Map<TopicPartition,OffsetAndMetadata> offsets, OffsetCommitCallback callback) Commit the specified offsets for the specified list of topics and partitions to Kafka.
void	commitAsync(OffsetCommitCallback callback) Commit offsets returned on the last poll() for the subscribed list of topics and partitions.
void	commitSync() Commit offsets returned on the last poll() for all the subscribed list of topics and partitions.
void	commitSync(java.util.Map<TopicPartition,OffsetAndMetadata> offsets) Commit the specified offsets for the specified list of topics and partitions.
OffsetAndMetadata	committed(TopicPartition partition) Get the last committed offset for the given partition (whether the commit happened by this process or another).
java.util.Map<java.lang.String,java.util.List<PartitionInfo>>	listTopics() Get metadata about partitions for all topics that the user is authorized to view.
java.util.Map<MetricName,? extends Metric>	metrics() Get the metrics kept by the consumer
java.util.List<PartitionInfo>	partitionsFor(java.lang.String topic) Get metadata about the partitions for a given topic.
void	pause(TopicPartition... partitions) Suspend fetching from the requested partitions.
ConsumerRecords<K,V>	poll(long timeout) Fetch data for the topics or partitions specified using one of the subscribe/assign APIs.
long	position(TopicPartition partition) Get the offset of the next record that will be fetched (if a record with that offset exists).
void	resume(TopicPartition... partitions) Resume specified partitions which have been paused with pause(TopicPartition...).
void	seek(TopicPartition partition, long offset) Overrides the fetch offsets that the consumer will use on the next poll(timeout).
void	seekToBeginning(TopicPartition... partitions) Seek to the first offset for each of the given partitions.
void	seekToEnd(TopicPartition... partitions) Seek to the last offset for each of the given partitions.
void	subscribe(java.util.List<java.lang.String> topics) Subscribe to the given list of topics to get dynamically assigned partitions.
void	subscribe(java.util.List<java.lang.String> topics, ConsumerRebalanceListener listener) Subscribe to the given list of topics to get dynamically assigned partitions.
void	subscribe(java.util.regex.Pattern pattern, ConsumerRebalanceListener listener) Subscribe to all topics matching specified pattern to get dynamically assigned partitions.
java.util.Set<java.lang.String>	subscription() Get the current subscription.
void	unsubscribe() Unsubscribe from topics currently subscribed with subscribe(List).
void	wakeup() Wakeup the consumer.

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaConsumer

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

A consumer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings are documented here. Values can be either strings or objects of the appropriate type (for example a numeric configuration would accept either the string "42" or the integer 42).

Valid configuration strings are documented at ConsumerConfig

Parameters:

configs - The consumer configs

KafkaConsumer

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs,
             Deserializer<K> keyDeserializer,
             Deserializer<V> valueDeserializer)

A consumer is instantiated by providing a set of key-value pairs as configuration, a ConsumerRebalanceListener implementation, a key and a value Deserializer.

Valid configuration strings are documented at ConsumerConfig

Parameters:

configs - The consumer configs

keyDeserializer - The deserializer for key that implements Deserializer. The configure() method won't be called in the consumer when the deserializer is passed in directly.

valueDeserializer - The deserializer for value that implements Deserializer. The configure() method won't be called in the consumer when the deserializer is passed in directly.

KafkaConsumer

public KafkaConsumer(java.util.Properties properties)

A consumer is instantiated by providing a Properties object as configuration. Valid configuration strings are documented at ConsumerConfig A consumer is instantiated by providing a Properties object as configuration. Valid configuration strings are documented at ConsumerConfig

KafkaConsumer

public KafkaConsumer(java.util.Properties properties,
             Deserializer<K> keyDeserializer,
             Deserializer<V> valueDeserializer)

A consumer is instantiated by providing a Properties object as configuration and a ConsumerRebalanceListener implementation, a key and a value Deserializer.

Valid configuration strings are documented at ConsumerConfig

Parameters:

properties - The consumer configuration properties

keyDeserializer - The deserializer for key that implements Deserializer. The configure() method won't be called in the consumer when the deserializer is passed in directly.

valueDeserializer - The deserializer for value that implements Deserializer. The configure() method won't be called in the consumer when the deserializer is passed in directly.

Method Detail

assignment

public java.util.Set<TopicPartition> assignment()

Get the set of partitions currently assigned to this consumer. If subscription happened by directly assigning partitions using assign(List) then this will simply return the same partitions that were assigned. If topic subscription was used, then this will give the set of topic partitions currently assigned to the consumer (which may be none if the assignment hasn't happened yet, or the partitions are in the process of getting reassigned).

Specified by:

assignment in interface Consumer<K,V>

Returns:

The set of partitions currently assigned to this consumer

See Also:

assignment()

subscription

public java.util.Set<java.lang.String> subscription()

Get the current subscription. Will return the same topics used in the most recent call to subscribe(List, ConsumerRebalanceListener), or an empty set if no such call has been made.

Specified by:

subscription in interface Consumer<K,V>

Returns:

The set of topics currently subscribed to

See Also:

subscription()

subscribe

public void subscribe(java.util.List<java.lang.String> topics,
             ConsumerRebalanceListener listener)

Subscribe to the given list of topics to get dynamically assigned partitions. Topic subscriptions are not incremental. This list will replace the current assignment (if there is one). Note that it is not possible to combine topic subscription with group management with manual partition assignment through assign(List). If the given list of topics is empty, it is treated the same as unsubscribe().

As part of group management, the consumer will keep track of the list of consumers that belong to a particular group and will trigger a rebalance operation if one of the following events trigger -

• Number of partitions change for any of the subscribed list of topics
• Topic is created or deleted
• An existing member of the consumer group dies
• A new member is added to an existing consumer group via the join API

When any of these events are triggered, the provided listener will be invoked first to indicate that the consumer's assignment has been revoked, and then again when the new assignment has been received. Note that this listener will immediately override any listener set in a previous call to subscribe. It is guaranteed, however, that the partitions revoked/assigned through this interface are from topics subscribed in this call. See ConsumerRebalanceListener for more details.

Specified by:

subscribe in interface Consumer<K,V>

Parameters:

topics - The list of topics to subscribe to

listener - Non-null listener instance to get notifications on partition assignment/revocation for the subscribed topics

See Also:

subscribe(List, ConsumerRebalanceListener)

subscribe

public void subscribe(java.util.List<java.lang.String> topics)

Subscribe to the given list of topics to get dynamically assigned partitions. Topic subscriptions are not incremental. This list will replace the current assignment (if there is one). It is not possible to combine topic subscription with group management with manual partition assignment through assign(List). If the given list of topics is empty, it is treated the same as unsubscribe().

This is a short-hand for subscribe(List, ConsumerRebalanceListener), which uses a noop listener. If you need the ability to either seek to particular offsets, you should prefer subscribe(List, ConsumerRebalanceListener), since group rebalances will cause partition offsets to be reset. You should also prefer to provide your own listener if you are doing your own offset management since the listener gives you an opportunity to commit offsets before a rebalance finishes.

Specified by:

subscribe in interface Consumer<K,V>

Parameters:

topics - The list of topics to subscribe to

See Also:

subscribe(List)

subscribe

public void subscribe(java.util.regex.Pattern pattern,
             ConsumerRebalanceListener listener)

Subscribe to all topics matching specified pattern to get dynamically assigned partitions. The pattern matching will be done periodically against topics existing at the time of check.

As part of group management, the consumer will keep track of the list of consumers that belong to a particular group and will trigger a rebalance operation if one of the following events trigger -

• Number of partitions change for any of the subscribed list of topics
• Topic is created or deleted
• An existing member of the consumer group dies
• A new member is added to an existing consumer group via the join API

Specified by:

subscribe in interface Consumer<K,V>

Parameters:

pattern - Pattern to subscribe to

See Also:

subscribe(Pattern, ConsumerRebalanceListener)

unsubscribe

public void unsubscribe()

Unsubscribe from topics currently subscribed with subscribe(List). This also clears any partitions directly assigned through assign(List).

Specified by:

unsubscribe in interface Consumer<K,V>

See Also:

unsubscribe()

assign

public void assign(java.util.List<TopicPartition> partitions)

Manually assign a list of partition to this consumer. This interface does not allow for incremental assignment and will replace the previous assignment (if there is one).

Manual topic assignment through this method does not use the consumer's group management functionality. As such, there will be no rebalance operation triggered when group membership or cluster and topic metadata change. Note that it is not possible to use both manual partition assignment with assign(List) and group assignment with subscribe(List, ConsumerRebalanceListener).

Specified by:

assign in interface Consumer<K,V>

Parameters:

partitions - The list of partitions to assign this consumer

See Also:

assign(List)

poll

public ConsumerRecords<K,V> poll(long timeout)

Fetch data for the topics or partitions specified using one of the subscribe/assign APIs. It is an error to not have subscribed to any topics or partitions before polling for data.

On each poll, consumer will try to use the last consumed offset as the starting offset and fetch sequentially. The last consumed offset can be manually set through seek(TopicPartition, long) or automatically set as the last committed offset for the subscribed list of partitions

Specified by:

poll in interface Consumer<K,V>

Parameters:

timeout - The time, in milliseconds, spent waiting in poll if data is not available. If 0, returns immediately with any records that are available now. Must not be negative.

Returns:

map of topic to records since the last fetch for the subscribed list of topics and partitions

Throws:

InvalidOffsetException - if the offset for a partition or set of partitions is undefined or out of range and no offset reset policy has been configured

WakeupException - if wakeup() is called before or while this function is called

AuthorizationException - if caller does Read access to any of the subscribed topics or to the configured groupId

KafkaException - for any other unrecoverable errors (e.g. invalid groupId or session timeout, errors deserializing key/value pairs, or any new error cases in future versions)

See Also:

poll(long)

commitSync

public void commitSync()

Commit offsets returned on the last poll() for all the subscribed list of topics and partitions.

This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used.

This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is encountered (in which case it is thrown to the caller).

Specified by:

commitSync in interface Consumer<K,V>

Throws:

CommitFailedException - if the commit failed and cannot be retried. This can only occur if you are using automatic group management with subscribe(List), or if there is an active group with the same groupId which is using group management.

WakeupException - if wakeup() is called before or while this function is called

AuthorizationException - if not authorized to the topic or to the configured groupId

KafkaException - for any other unrecoverable errors (e.g. if offset metadata is too large or if the committed offset is invalid).

See Also:

commitSync()

commitSync

public void commitSync(java.util.Map<TopicPartition,OffsetAndMetadata> offsets)

Commit the specified offsets for the specified list of topics and partitions.

This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used. The committed offset should be the next message your application will consume, i.e. lastProcessedMessageOffset + 1.

This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is encountered (in which case it is thrown to the caller).

Specified by:

commitSync in interface Consumer<K,V>

Parameters:

offsets - A map of offsets by partition with associated metadata

Throws:

CommitFailedException - if the commit failed and cannot be retried. This can only occur if you are using automatic group management with subscribe(List), or if there is an active group with the same groupId which is using group management.

WakeupException - if wakeup() is called before or while this function is called

AuthorizationException - if not authorized to the topic or to the configured groupId

KafkaException - for any other unrecoverable errors (e.g. if offset metadata is too large or if the committed offset is invalid).

See Also:

commitSync(Map)

commitAsync

public void commitAsync()

Commit offsets returned on the last poll() for all the subscribed list of topics and partition. Same as commitAsync(null)

Specified by:

commitAsync in interface Consumer<K,V>

See Also:

commitAsync()

commitAsync

public void commitAsync(OffsetCommitCallback callback)

Commit offsets returned on the last poll() for the subscribed list of topics and partitions.

This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used.

This is an asynchronous call and will not block. Any errors encountered are either passed to the callback (if provided) or discarded.

Specified by:

commitAsync in interface Consumer<K,V>

Parameters:

callback - Callback to invoke when the commit completes

See Also:

commitAsync(OffsetCommitCallback)

commitAsync

public void commitAsync(java.util.Map<TopicPartition,OffsetAndMetadata> offsets,
               OffsetCommitCallback callback)

Commit the specified offsets for the specified list of topics and partitions to Kafka.

This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API should not be used. The committed offset should be the next message your application will consume, i.e. lastProcessedMessageOffset + 1.

This is an asynchronous call and will not block. Any errors encountered are either passed to the callback (if provided) or discarded.

Specified by:

commitAsync in interface Consumer<K,V>

Parameters:

offsets - A map of offsets by partition with associate metadata. This map will be copied internally, so it is safe to mutate the map after returning.

callback - Callback to invoke when the commit completes

See Also:

commitAsync(Map, OffsetCommitCallback)

seek

public void seek(TopicPartition partition,
        long offset)

Overrides the fetch offsets that the consumer will use on the next poll(timeout). If this API is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets

Specified by:

seek in interface Consumer<K,V>

See Also:

seek(TopicPartition, long)

seekToBeginning

public void seekToBeginning(TopicPartition... partitions)

Seek to the first offset for each of the given partitions. This function evaluates lazily, seeking to the final offset in all partitions only when poll(long) or position(TopicPartition) are called.

Specified by:

seekToBeginning in interface Consumer<K,V>

See Also:

seekToBeginning(TopicPartition...)

seekToEnd

public void seekToEnd(TopicPartition... partitions)

Seek to the last offset for each of the given partitions. This function evaluates lazily, seeking to the final offset in all partitions only when poll(long) or position(TopicPartition) are called.

Specified by:

seekToEnd in interface Consumer<K,V>

See Also:

seekToEnd(TopicPartition...)

position

public long position(TopicPartition partition)

Get the offset of the next record that will be fetched (if a record with that offset exists).

Specified by:

position in interface Consumer<K,V>

Parameters:

partition - The partition to get the position for

Returns:

The offset

Throws:

InvalidOffsetException - if no offset is currently defined for the partition

WakeupException - if wakeup() is called before or while this function is called

AuthorizationException - if not authorized to the topic or to the configured groupId

KafkaException - for any other unrecoverable errors

See Also:

position(TopicPartition)

committed

public OffsetAndMetadata committed(TopicPartition partition)

Get the last committed offset for the given partition (whether the commit happened by this process or another). This offset will be used as the position for the consumer in the event of a failure.

This call may block to do a remote call if the partition in question isn't assigned to this consumer or if the consumer hasn't yet initialized its cache of committed offsets.

Specified by:

committed in interface Consumer<K,V>

Parameters:

partition - The partition to check

Returns:

The last committed offset and metadata or null if there was no prior commit

Throws:

WakeupException - if wakeup() is called before or while this function is called

AuthorizationException - if not authorized to the topic or to the configured groupId

KafkaException - for any other unrecoverable errors

See Also:

committed(TopicPartition)

metrics

public java.util.Map<MetricName,? extends Metric> metrics()

Get the metrics kept by the consumer

Specified by:

metrics in interface Consumer<K,V>

See Also:

metrics()

partitionsFor

public java.util.List<PartitionInfo> partitionsFor(java.lang.String topic)

Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it does not already have any metadata about the given topic.

Specified by:

partitionsFor in interface Consumer<K,V>

Parameters:

topic - The topic to get partition metadata for

Returns:

The list of partitions

Throws:

WakeupException - if wakeup() is called before or while this function is called

AuthorizationException - if not authorized to the specified topic

TimeoutException - if the topic metadata could not be fetched before expiration of the configured request timeout

KafkaException - for any other unrecoverable errors

See Also:

partitionsFor(String)

listTopics

public java.util.Map<java.lang.String,java.util.List<PartitionInfo>> listTopics()

Get metadata about partitions for all topics that the user is authorized to view. This method will issue a remote call to the server.

Specified by:

listTopics in interface Consumer<K,V>

Returns:

The map of topics and its partitions

Throws:

WakeupException - if wakeup() is called before or while this function is called

TimeoutException - if the topic metadata could not be fetched before expiration of the configured request timeout

KafkaException - for any other unrecoverable errors

See Also:

listTopics()

pause

public void pause(TopicPartition... partitions)

Suspend fetching from the requested partitions. Future calls to poll(long) will not return any records from these partitions until they have been resumed using resume(TopicPartition...). Note that this method does not affect partition subscription. In particular, it does not cause a group rebalance when automatic assignment is used.

Specified by:

pause in interface Consumer<K,V>

Parameters:

partitions - The partitions which should be paused

See Also:

pause(TopicPartition...)

resume

public void resume(TopicPartition... partitions)

Resume specified partitions which have been paused with pause(TopicPartition...). New calls to poll(long) will return records from these partitions if there are any to be fetched. If the partitions were not previously paused, this method is a no-op.

Specified by:

resume in interface Consumer<K,V>

Parameters:

partitions - The partitions which should be resumed

See Also:

resume(TopicPartition...)

close

public void close()

Close the consumer, waiting indefinitely for any needed cleanup. If auto-commit is enabled, this will commit the current offsets. Note that wakeup() cannot be use to interrupt close.

Specified by:

close in interface java.io.Closeable

Specified by:

close in interface java.lang.AutoCloseable

Specified by:

close in interface Consumer<K,V>

See Also:

close()

wakeup

public void wakeup()

Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll. The thread which is blocking in an operation will throw WakeupException.

Specified by:

wakeup in interface Consumer<K,V>

See Also:

wakeup()