Class KafkaConsumer<K,V>
- All Implemented Interfaces:
Closeable
,AutoCloseable
,Consumer<K,
V>
This client transparently handles the failure of Kafka brokers, and transparently adapts as topic partitions it fetches migrate within the cluster. This client also interacts with the broker to allow groups of consumers to load balance consumption using consumer groups.
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.
Cross-Version Compatibility
This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support certain features. For example, 0.10.0 brokers do not support offsetsForTimes, because this feature was added in version 0.10.1. You will receive anUnsupportedVersionException
when invoking an API that is not available on the running broker version.
Offsets and Consumer Position
Kafka maintains a numerical offset for each record in a partition. This offset acts as a unique identifier of a record within that partition, and also denotes the position of the consumer in the partition. For example, a consumer which is at position 5 has consumed records with offsets 0 through 4 and will next receive the 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 messages in a call to poll(Duration)
.
The committed position
is the last offset that has been stored securely. Should the
process fail and restart, this is the offset that the consumer will recover to. The consumer can either automatically commit
offsets periodically; or it can choose to control this committed position manually by calling one of the commit APIs
(e.g. commitSync
and commitAsync
).
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 the work of consuming and processing records. These processes can either be running on the same machine or they can be distributed over many machines to provide scalability and fault tolerance for processing. All consumer instances sharing the samegroup.id
will be part of the same consumer group.
Each consumer in a group can dynamically set the list of topics it wants to subscribe to through one of the
subscribe
APIs. Kafka will deliver each message in the
subscribed topics to one process in each consumer group. This is achieved by balancing the partitions between all
members in the consumer group so that each partition is assigned to exactly one consumer in the group. So if there
is a topic with four partitions, and a consumer group with two processes, each process would consume from two partitions.
Membership in a consumer group is maintained dynamically: if a process fails, the partitions assigned to it will
be reassigned to other consumers in the same group. Similarly, if a new consumer joins the group, partitions will be moved
from existing consumers to the new one. This is known as rebalancing the group and is discussed in more
detail below. Group rebalancing is also used when new partitions are added
to one of the subscribed topics or when a new topic matching a subscribed regex
is created. The group will automatically detect the new partitions through periodic metadata refreshes and
assign them to members of the group.
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 a ConsumerRebalanceListener
,
which allows them to finish necessary application-level logic such as state cleanup, manual offset
commits, etc. See Storing Offsets Outside Kafka for more details.
It is also possible for the consumer to manually assign specific partitions
(similar to the older "simple" consumer) using assign(Collection)
. In this case, dynamic partition
assignment and consumer group coordination will be disabled.
Detecting Consumer Failures
After subscribing to a set of topics, the consumer will automatically join the group whenpoll(Duration)
is
invoked. The poll API is designed to ensure consumer liveness. As long as you continue to call poll, the consumer
will stay in the group and continue to receive messages from the partitions it was assigned. Underneath the covers,
the consumer sends periodic heartbeats to the server. If the consumer crashes or is unable to send heartbeats for
a duration of session.timeout.ms
, then the consumer will be considered dead and its partitions will
be reassigned.
It is also possible that the consumer could encounter a "livelock" situation where it is continuing
to send heartbeats, but no progress is being made. To prevent the consumer from holding onto its partitions
indefinitely in this case, we provide a liveness detection mechanism using the max.poll.interval.ms
setting. Basically if you don't call poll at least as frequently as the configured max interval,
then the client will proactively leave the group so that another consumer can take over its partitions. When this happens,
you may see an offset commit failure (as indicated by a CommitFailedException
thrown from a call to commitSync()
).
This is a safety mechanism which guarantees that only active members of the group are able to commit offsets.
So to stay in the group, you must continue to call poll.
The consumer provides two configuration settings to control the behavior of the poll loop:
max.poll.interval.ms
: By increasing the interval between expected polls, you can give the consumer more time to handle a batch of records returned frompoll(Duration)
. The drawback is that increasing this value may delay a group rebalance since the consumer will only join the rebalance inside the call to poll. You can use this setting to bound the time to finish a rebalance, but you risk slower progress if the consumer cannot actually callpoll
often enough.max.poll.records
: Use this setting to limit the total records returned from a single call to poll. This can make it easier to predict the maximum that must be handled within each poll interval. By tuning this value, you may be able to reduce the poll interval, which will reduce the impact of group rebalancing.
For use cases where message processing time varies unpredictably, neither of these options may be sufficient.
The recommended way to handle these cases is to move message processing to another thread, which allows
the consumer to continue calling poll
while the processor is still working.
Some care must be taken to ensure that committed offsets do not get ahead of the actual position.
Typically, you must disable automatic commits and manually commit processed offsets for records only after the
thread has finished handling them (depending on the delivery semantics you need).
Note also that you will need to pause
the partition so that no new records are received
from poll until after thread has finished handling those previously returned.
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 relies on automatic offset committing.
Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "test"); props.setProperty("enable.auto.commit", "true"); props.setProperty("auto.commit.interval.ms", "1000"); props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("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(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); }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).
Setting enable.auto.commit
means that offsets are committed automatically with a frequency controlled by
the config auto.commit.interval.ms
.
In this example the consumer is subscribing to the topics foo and bar as part of a group of consumers
called test as configured with group.id
.
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 records should be considered as consumed and hence commit their offsets. This is useful when the consumption of the messages is coupled with some processing logic and hence a message should not be considered as consumed until it is completed processing.
Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "test"); props.setProperty("enable.auto.commit", "false"); props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("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(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { buffer.add(record); } if (buffer.size() >= minBatchSize) { insertIntoDb(buffer); consumer.commitSync(); buffer.clear(); } }In this example we will consume a batch of records and batch them up in memory. When we have enough records batched, we will insert them into a database. If we allowed offsets to auto commit as in the previous example, records would be considered consumed after they were returned to the user in
poll
. It would then be
possible
for our process to fail after batching the records, but before they had been inserted into the database.
To avoid this, we will manually commit the offsets only after the corresponding records have been inserted into the database. This gives us exact control of when a record 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 record will likely be delivered one time but in failure cases could be duplicated.
Note: Using automatic offset commits can also give you "at-least-once" delivery, but the requirement is that
you must consume all data returned from each call to poll(Duration)
before any subsequent calls, or before
closing
the consumer. If you fail to do either of these, it is possible for the committed offset
to get ahead of the consumed position, which results in missing records. The advantage of using manual offset
control is that you have direct control over when a record is considered "consumed."
The above example uses commitSync
to mark all received records as committed. In some cases
you may wish to have even finer control over which records have been committed by specifying an offset explicitly.
In the example below we commit offset after we finish handling the records in each partition.
try { while(running) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(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.
Manual Partition Assignment
In the previous examples, we subscribed to the topics we were interested in and let Kafka dynamically assign a fair share of the partitions for those topics based on the active consumers in the group. However, in some cases you may need finer control over the specific partitions that are assigned. For example:
- If the process is maintaining some kind of local state associated with that partition (like a local on-disk key-value store), then it should only get records for the partition it is maintaining on disk.
- 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 since the consuming process will be restarted on another machine.
To use this mode, instead of subscribing to the topic using subscribe
, you just call
assign(Collection)
with the full list of partitions that you want to consume.
String topic = "foo"; TopicPartition partition0 = new TopicPartition(topic, 0); TopicPartition partition1 = new TopicPartition(topic, 1); consumer.assign(Arrays.asList(partition0, partition1));Once assigned, you can call
poll
in a loop, just as in the preceding examples to consume
records. The group that the consumer specifies is still used for committing offsets, but now the set of partitions
will only change with another call to assign
. Manual partition assignment does
not use group coordination, so consumer failures will not cause assigned partitions to be rebalanced. Each consumer
acts independently even if it shares a groupId with another consumer. To avoid offset commit conflicts, you should
usually ensure that the groupId is unique for each consumer instance.
Note that it isn't possible to mix manual partition assignment (i.e. using assign
)
with dynamic partition assignment through topic subscription (i.e. using subscribe
).
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(Collection, 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(Collection)
and seekToEnd(Collection)
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 topics 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 bootstrapping upon consumer starting up where there are a lot of history data to catch up, the applications usually want 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(Collection)
and resume(Collection)
to pause the consumption on the specified assigned partitions and resume the consumption
on the specified paused partitions respectively in the future poll(Duration)
calls.
Reading Transactional Messages
Transactions were introduced in Kafka 0.11.0 wherein applications can write to multiple topics and partitions atomically.
In order for this to work, consumers reading from these partitions should be configured to only read committed data.
This can be achieved by setting the isolation.level=read_committed
in the consumer's configuration.
In read_committed
mode, the consumer will read only those transactional messages which have been
successfully committed. It will continue to read non-transactional messages as before. There is no client-side
buffering in read_committed
mode. Instead, the end offset of a partition for a read_committed
consumer would be the offset of the first message in the partition belonging to an open transaction. This offset
is known as the 'Last Stable Offset'(LSO).
A read_committed
consumer will only read up to the LSO and filter out any transactional
messages which have been aborted. The LSO also affects the behavior of seekToEnd(Collection)
and
endOffsets(Collection)
for read_committed
consumers, details of which are in each method's documentation.
Finally, the fetch lag metrics are also adjusted to be relative to the LSO for read_committed
consumers.
Partitions with transactional messages will include commit or abort markers which indicate the result of a transaction.
There markers are not returned to applications, yet have an offset in the log. As a result, applications reading from
topics with transactional messages will see gaps in the consumed offsets. These missing messages would be the transaction
markers, and they are filtered out for consumers in both isolation levels. Additionally, applications using
read_committed
consumers may also see gaps due to aborted transactions, since those messages would not
be returned by the consumer and yet would have valid offsets.
Multi-threaded Processing
The Kafka consumer is NOT thread-safe. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized. Un-synchronized access will result inConcurrentModificationException
.
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 KafkaConsumerRunner(KafkaConsumer consumer) { this.consumer = consumer; } @Override public void run() { try { consumer.subscribe(Arrays.asList("topic")); while (!closed.get()) { ConsumerRecords records = consumer.poll(Duration.ofMillis(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();
Note that while it is possible to use thread interrupts instead of wakeup()
to abort a blocking operation
(in which case, InterruptException
will be raised), we discourage their use since they may cause a clean
shutdown of the consumer to be aborted. Interrupts are mainly supported for those cases where using wakeup()
is impossible, e.g. when a consumer thread is managed by code that is unaware of the Kafka client.
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 offConsumerRecords
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.
-
Constructor Summary
ConstructorDescriptionKafkaConsumer
(Map<String, Object> configs) A consumer is instantiated by providing a set of key-value pairs as configuration.KafkaConsumer
(Map<String, Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a valueDeserializer
.KafkaConsumer
(Properties properties) A consumer is instantiated by providing aProperties
object as configuration.KafkaConsumer
(Properties properties, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) A consumer is instantiated by providing aProperties
object as configuration, and a key and a valueDeserializer
. -
Method Summary
Modifier and TypeMethodDescriptionvoid
assign
(Collection<TopicPartition> partitions) Manually assign a list of partitions to this consumer.Get the set of partitions currently assigned to this consumer.beginningOffsets
(Collection<TopicPartition> partitions) Get the first offset for the given partitions.beginningOffsets
(Collection<TopicPartition> partitions, Duration timeout) Get the first offset for the given partitions.clientInstanceId
(Duration timeout) Determines the client's unique client instance ID used for telemetry.void
close()
Close the consumer, waiting for up to the default timeout of 30 seconds for any needed cleanup.void
Tries to close the consumer cleanly within the specified timeout.void
Commit offsets returned on the lastpoll(Duration)
for all the subscribed list of topics and partition.void
commitAsync
(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 lastpoll()
for the subscribed list of topics and partitions.void
Commit offsets returned on the lastpoll()
for all the subscribed list of topics and partitions.void
commitSync
(Duration timeout) Commit offsets returned on the lastpoll()
for all the subscribed list of topics and partitions.void
commitSync
(Map<TopicPartition, OffsetAndMetadata> offsets) Commit the specified offsets for the specified list of topics and partitions.void
commitSync
(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout) Commit the specified offsets for the specified list of topics and partitions.committed
(Set<TopicPartition> partitions) Get the last committed offsets for the given partitions (whether the commit happened by this process or another).committed
(Set<TopicPartition> partitions, Duration timeout) Get the last committed offsets for the given partitions (whether the commit happened by this process or another).committed
(TopicPartition partition) Deprecated.committed
(TopicPartition partition, Duration timeout) Deprecated.since 2.4 Usecommitted(Set, Duration)
insteadcurrentLag
(TopicPartition topicPartition) Get the consumer's current lag on the partition.endOffsets
(Collection<TopicPartition> partitions) Get the end offsets for the given partitions.endOffsets
(Collection<TopicPartition> partitions, Duration timeout) Get the end offsets for the given partitions.void
void
enforceRebalance
(String reason) Alert the consumer to trigger a new rebalance by rejoining the group.Return the current group metadata associated with this consumer.Get metadata about partitions for all topics that the user is authorized to view.listTopics
(Duration timeout) Get metadata about partitions for all topics that the user is authorized to view.Map<MetricName,
? extends Metric> metrics()
Get the metrics kept by the consumeroffsetsForTimes
(Map<TopicPartition, Long> timestampsToSearch) Look up the offsets for the given partitions by timestamp.offsetsForTimes
(Map<TopicPartition, Long> timestampsToSearch, Duration timeout) Look up the offsets for the given partitions by timestamp.partitionsFor
(String topic) Get metadata about the partitions for a given topic.partitionsFor
(String topic, Duration timeout) Get metadata about the partitions for a given topic.void
pause
(Collection<TopicPartition> partitions) Suspend fetching from the requested partitions.paused()
Get the set of partitions that were previously paused by a call topause(Collection)
.poll
(long timeoutMs) Deprecated.Since 2.0.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).long
position
(TopicPartition partition, Duration timeout) Get the offset of the next record that will be fetched (if a record with that offset exists).void
resume
(Collection<TopicPartition> partitions) Resume specified partitions which have been paused withpause(Collection)
.void
seek
(TopicPartition partition, long offset) Overrides the fetch offsets that the consumer will use on the nextpoll(timeout)
.void
seek
(TopicPartition partition, OffsetAndMetadata offsetAndMetadata) Overrides the fetch offsets that the consumer will use on the nextpoll(timeout)
.void
seekToBeginning
(Collection<TopicPartition> partitions) Seek to the first offset for each of the given partitions.void
seekToEnd
(Collection<TopicPartition> partitions) Seek to the last offset for each of the given partitions.void
subscribe
(Collection<String> topics) Subscribe to the given list of topics to get dynamically assigned partitions.void
subscribe
(Collection<String> topics, ConsumerRebalanceListener listener) Subscribe to the given list of topics to get dynamically assigned partitions.void
Subscribe to all topics matching specified pattern to get dynamically assigned partitions.void
subscribe
(Pattern pattern, ConsumerRebalanceListener listener) Subscribe to all topics matching specified pattern to get dynamically assigned partitions.Get the current subscription.void
Unsubscribe from topics currently subscribed withsubscribe(Collection)
orsubscribe(Pattern)
.void
wakeup()
Wakeup the consumer.
-
Constructor Details
-
KafkaConsumer
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
.Note: after creating a
KafkaConsumer
you must alwaysclose()
it to avoid resource leaks.- Parameters:
configs
- The consumer configs
-
KafkaConsumer
A consumer is instantiated by providing aProperties
object as configuration.Valid configuration strings are documented at
ConsumerConfig
.Note: after creating a
KafkaConsumer
you must alwaysclose()
it to avoid resource leaks.- Parameters:
properties
- The consumer configuration properties
-
KafkaConsumer
public KafkaConsumer(Properties properties, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) A consumer is instantiated by providing aProperties
object as configuration, and a key and a valueDeserializer
.Valid configuration strings are documented at
ConsumerConfig
.Note: after creating a
KafkaConsumer
you must alwaysclose()
it to avoid resource leaks.- Parameters:
properties
- The consumer configuration propertieskeyDeserializer
- The deserializer for key that implementsDeserializer
. The configure() method won't be called in the consumer when the deserializer is passed in directly.valueDeserializer
- The deserializer for value that implementsDeserializer
. The configure() method won't be called in the consumer when the deserializer is passed in directly.
-
KafkaConsumer
public KafkaConsumer(Map<String, Object> configs, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a valueDeserializer
.Valid configuration strings are documented at
ConsumerConfig
.Note: after creating a
KafkaConsumer
you must alwaysclose()
it to avoid resource leaks.- Parameters:
configs
- The consumer configskeyDeserializer
- The deserializer for key that implementsDeserializer
. The configure() method won't be called in the consumer when the deserializer is passed in directly.valueDeserializer
- The deserializer for value that implementsDeserializer
. The configure() method won't be called in the consumer when the deserializer is passed in directly.
-
-
Method Details
-
assignment
Get the set of partitions currently assigned to this consumer. If subscription happened by directly assigning partitions usingassign(Collection)
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 interfaceConsumer<K,
V> - Returns:
- The set of partitions currently assigned to this consumer
- See Also:
-
subscription
Get the current subscription. Will return the same topics used in the most recent call tosubscribe(Collection, ConsumerRebalanceListener)
, or an empty set if no such call has been made.- Specified by:
subscription
in interfaceConsumer<K,
V> - Returns:
- The set of topics currently subscribed to
- See Also:
-
subscribe
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 throughassign(Collection)
. If the given list of topics is empty, it is treated the same asunsubscribe()
.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 any one of the following events are triggered:
- Number of partitions change for any of the subscribed topics
- A subscribed topic is created or deleted
- An existing member of the consumer group is shutdown or fails
- A new member is added to the consumer group
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 rebalances will only occur during an active call to
poll(Duration)
, so callbacks will also only be invoked during that time. The provided 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. SeeConsumerRebalanceListener
for more details.- Specified by:
subscribe
in interfaceConsumer<K,
V> - Parameters:
topics
- The list of topics to subscribe tolistener
- Non-null listener instance to get notifications on partition assignment/revocation for the subscribed topics- Throws:
IllegalArgumentException
- If topics is null or contains null or empty elements, or if listener is nullIllegalStateException
- Ifsubscribe()
is called previously with pattern, or assign is called previously (without a subsequent call tounsubscribe()
), or if not configured at-least one partition assignment strategy- See Also:
-
subscribe
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 throughassign(Collection)
. If the given list of topics is empty, it is treated the same asunsubscribe()
.This is a short-hand for
subscribe(Collection, ConsumerRebalanceListener)
, which uses a no-op listener. If you need the ability to seek to particular offsets, you should prefersubscribe(Collection, ConsumerRebalanceListener)
, since group rebalances will cause partition offsets to be reset. You should also 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 interfaceConsumer<K,
V> - Parameters:
topics
- The list of topics to subscribe to- Throws:
IllegalArgumentException
- If topics is null or contains null or empty elementsIllegalStateException
- Ifsubscribe()
is called previously with pattern, or assign is called previously (without a subsequent call tounsubscribe()
), or if not configured at-least one partition assignment strategy- See Also:
-
subscribe
Subscribe to all topics matching specified pattern to get dynamically assigned partitions. The pattern matching will be done periodically against all topics existing at the time of check. This can be controlled through themetadata.max.age.ms
configuration: by lowering the max metadata age, the consumer will refresh metadata more often and check for matching topics.See
subscribe(Collection, ConsumerRebalanceListener)
for details on the use of theConsumerRebalanceListener
. Generally rebalances are triggered when there is a change to the topics matching the provided pattern and when consumer group membership changes. Group rebalances only take place during an active call topoll(Duration)
.- Specified by:
subscribe
in interfaceConsumer<K,
V> - Parameters:
pattern
- Pattern to subscribe tolistener
- Non-null listener instance to get notifications on partition assignment/revocation for the subscribed topics- Throws:
IllegalArgumentException
- If pattern or listener is nullIllegalStateException
- Ifsubscribe()
is called previously with topics, or assign is called previously (without a subsequent call tounsubscribe()
), or if not configured at-least one partition assignment strategy- See Also:
-
subscribe
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.This is a short-hand for
subscribe(Pattern, ConsumerRebalanceListener)
, which uses a no-op listener. If you need the ability to seek to particular offsets, you should prefersubscribe(Pattern, ConsumerRebalanceListener)
, since group rebalances will cause partition offsets to be reset. You should also 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 interfaceConsumer<K,
V> - Parameters:
pattern
- Pattern to subscribe to- Throws:
IllegalArgumentException
- If pattern is nullIllegalStateException
- Ifsubscribe()
is called previously with topics, or assign is called previously (without a subsequent call tounsubscribe()
), or if not configured at-least one partition assignment strategy- See Also:
-
unsubscribe
public void unsubscribe()Unsubscribe from topics currently subscribed withsubscribe(Collection)
orsubscribe(Pattern)
. This also clears any partitions directly assigned throughassign(Collection)
.- Specified by:
unsubscribe
in interfaceConsumer<K,
V> - Throws:
KafkaException
- for any other unrecoverable errors (e.g. rebalance callback errors)- See Also:
-
assign
Manually assign a list of partitions to this consumer. This interface does not allow for incremental assignment and will replace the previous assignment (if there is one).If the given list of topic partitions is empty, it is treated the same as
unsubscribe()
.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(Collection)
and group assignment withsubscribe(Collection, ConsumerRebalanceListener)
.If auto-commit is enabled, an async commit (based on the old assignment) will be triggered before the new assignment replaces the old one.
- Specified by:
assign
in interfaceConsumer<K,
V> - Parameters:
partitions
- The list of partitions to assign this consumer- Throws:
IllegalArgumentException
- If partitions is null or contains null or empty topicsIllegalStateException
- Ifsubscribe()
is called previously with topics or pattern (without a subsequent call tounsubscribe()
)- See Also:
-
poll
Deprecated.Since 2.0. Usepoll(Duration)
, which does not block beyond the timeout awaiting partition assignment. See KIP-266 for more information.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 interfaceConsumer<K,
V> - Parameters:
timeoutMs
- The time, in milliseconds, spent waiting in poll if data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. 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 configuredWakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if caller lacks Read access to any of the subscribed topics or to the configured groupId. See the exception for more detailsKafkaException
- 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)IllegalArgumentException
- if the timeout value is negativeIllegalStateException
- if the consumer is not subscribed to any topics or manually assigned any partitions to consume fromFencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
poll
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 partitionsThis method returns immediately if there are records available or if the position advances past control records or aborted transactions when isolation.level=read_committed. Otherwise, it will await the passed timeout. If the timeout expires, an empty record set will be returned. Note that this method may block beyond the timeout in order to execute custom
ConsumerRebalanceListener
callbacks.- Specified by:
poll
in interfaceConsumer<K,
V> - Parameters:
timeout
- The maximum time to block (must not be greater thanLong.MAX_VALUE
milliseconds)- 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 configuredWakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if caller lacks Read access to any of the subscribed topics or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errors (e.g. invalid groupId or session timeout, errors deserializing key/value pairs, your rebalance callback thrown exceptions, or any new error cases in future versions)IllegalArgumentException
- if the timeout value is negativeIllegalStateException
- if the consumer is not subscribed to any topics or manually assigned any partitions to consume fromArithmeticException
- if the timeout is greater thanLong.MAX_VALUE
milliseconds.InvalidTopicException
- if the current subscription contains any invalid topic (perTopic.validate(String)
)UnsupportedVersionException
- if the consumer attempts to fetch stable offsets when the broker doesn't support this featureFencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitSync
public void commitSync()Commit offsets returned on the lastpoll()
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 commit and will block until either the commit succeeds, an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout specified by
default.api.timeout.ms
expires (in which case aTimeoutException
is thrown to the caller).Note that asynchronous offset commits sent previously with the
commitAsync(OffsetCommitCallback)
(or similar) are guaranteed to have their callbacks invoked prior to completion of this method.- Specified by:
commitSync
in interfaceConsumer<K,
V> - Throws:
CommitFailedException
- if the commit failed and cannot be retried. This fatal error can only occur if you are using automatic group management withsubscribe(Collection)
, or if there is an active group with the samegroup.id
which is using group management. In such cases, when you are trying to commit to partitions that are no longer assigned to this consumer because the consumer is for example no longer part of the group this exception would be thrown.RebalanceInProgressException
- if the consumer instance is in the middle of a rebalance so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first complete the rebalance by callingpoll(Duration)
and commit can be reconsidered afterwards. NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, and also for those partitions that are still assigned their fetch positions may have changed too if more records are returned from thepoll(Duration)
call.WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errors (e.g. if offset metadata is too large or if the topic does not exist).TimeoutException
- if the timeout specified bydefault.api.timeout.ms
expires before successful completion of the offset commitFencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitSync
Commit offsets returned on the lastpoll()
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 commit and will block until either the commit succeeds, an unrecoverable error is encountered (in which case it is thrown to the caller), or the passed timeout expires.
Note that asynchronous offset commits sent previously with the
commitAsync(OffsetCommitCallback)
(or similar) are guaranteed to have their callbacks invoked prior to completion of this method.- Specified by:
commitSync
in interfaceConsumer<K,
V> - Throws:
CommitFailedException
- if the commit failed and cannot be retried. This can only occur if you are using automatic group management withsubscribe(Collection)
, or if there is an active group with the samegroup.id
which is using group management. In such cases, when you are trying to commit to partitions that are no longer assigned to this consumer because the consumer is for example no longer part of the group this exception would be thrown.RebalanceInProgressException
- if the consumer instance is in the middle of a rebalance so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first complete the rebalance by callingpoll(Duration)
and commit can be reconsidered afterwards. NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, and also for those partitions that are still assigned their fetch positions may have changed too if more records are returned from thepoll(Duration)
call.WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errors (e.g. if offset metadata is too large or if the topic does not exist).TimeoutException
- if the timeout expires before successful completion of the offset commitFencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitSync
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. If automatic group management with
subscribe(Collection)
is used, then the committed offsets must belong to the currently auto-assigned partitions.This is a synchronous commit and will block until either the commit succeeds or an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout specified by
default.api.timeout.ms
expires (in which case aTimeoutException
is thrown to the caller).Note that asynchronous offset commits sent previously with the
commitAsync(OffsetCommitCallback)
(or similar) are guaranteed to have their callbacks invoked prior to completion of this method.- Specified by:
commitSync
in interfaceConsumer<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 withsubscribe(Collection)
, or if there is an active group with the samegroup.id
which is using group management. In such cases, when you are trying to commit to partitions that are no longer assigned to this consumer because the consumer is for example no longer part of the group this exception would be thrown.RebalanceInProgressException
- if the consumer instance is in the middle of a rebalance so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first complete the rebalance by callingpoll(Duration)
and commit can be reconsidered afterwards. NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, and also for those partitions that are still assigned their fetch positions may have changed too if more records are returned from thepoll(Duration)
call, so when you retry committing you should consider updating the passed inoffset
parameter.WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsIllegalArgumentException
- if the committed offset is negativeKafkaException
- for any other unrecoverable errors (e.g. if offset metadata is too large or if the topic does not exist).TimeoutException
- if the timeout expires before successful completion of the offset commitFencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitSync
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. If automatic group management with
subscribe(Collection)
is used, then the committed offsets must belong to the currently auto-assigned partitions.This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout expires.
Note that asynchronous offset commits sent previously with the
commitAsync(OffsetCommitCallback)
(or similar) are guaranteed to have their callbacks invoked prior to completion of this method.- Specified by:
commitSync
in interfaceConsumer<K,
V> - Parameters:
offsets
- A map of offsets by partition with associated metadatatimeout
- The maximum amount of time to await completion of the offset commit- Throws:
CommitFailedException
- if the commit failed and cannot be retried. This can only occur if you are using automatic group management withsubscribe(Collection)
, or if there is an active group with the samegroup.id
which is using group management. In such cases, when you are trying to commit to partitions that are no longer assigned to this consumer because the consumer is for example no longer part of the group this exception would be thrown.RebalanceInProgressException
- if the consumer instance is in the middle of a rebalance so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first complete the rebalance by callingpoll(Duration)
and commit can be reconsidered afterwards. NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed, and also for those partitions that are still assigned their fetch positions may have changed too if more records are returned from thepoll(Duration)
call, so when you retry committing you should consider updating the passed inoffset
parameter.WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsIllegalArgumentException
- if the committed offset is negativeKafkaException
- for any other unrecoverable errors (e.g. if offset metadata is too large or if the topic does not exist).TimeoutException
- if the timeout expires before successful completion of the offset commitFencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitAsync
public void commitAsync()Commit offsets returned on the lastpoll(Duration)
for all the subscribed list of topics and partition. Same ascommitAsync(null)
- Specified by:
commitAsync
in interfaceConsumer<K,
V> - Throws:
FencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitAsync
Commit offsets returned on the lastpoll()
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.
Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that offsets committed through this API are guaranteed to complete before a subsequent call to
commitSync()
(and variants) returns.- Specified by:
commitAsync
in interfaceConsumer<K,
V> - Parameters:
callback
- Callback to invoke when the commit completes- Throws:
FencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
commitAsync
public void commitAsync(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. If automatic group management with
subscribe(Collection)
is used, then the committed offsets must belong to the currently auto-assigned partitions.This is an asynchronous call and will not block. Any errors encountered are either passed to the callback (if provided) or discarded.
Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that offsets committed through this API are guaranteed to complete before a subsequent call to
commitSync()
(and variants) returns.- Specified by:
commitAsync
in interfaceConsumer<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- Throws:
FencedInstanceIdException
- if this consumer instance gets fenced by broker.- See Also:
-
seek
Overrides the fetch offsets that the consumer will use on the nextpoll(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 offsetsThe next Consumer Record which will be retrieved when poll() is invoked will have the offset specified, given that a record with that offset exists (i.e. it is a valid offset).
seekToBeginning(Collection)
will go to the first offset in the topic. seek(0) is equivalent to seekToBeginning for a TopicPartition with beginning offset 0, assuming that there is a record at offset 0 still available.seekToEnd(Collection)
is equivalent to seeking to the last offset of the partition, but behavior depends onisolation.level
, so seeseekToEnd(Collection)
documentation for more details.Seeking to the offset smaller than the log start offset or larger than the log end offset means an invalid offset is reached. Invalid offset behaviour is controlled by the
auto.offset.reset
property. If this is set to "earliest", the next poll will return records from the starting offset. If it is set to "latest", it will seek to the last offset (similar to seekToEnd()). If it is set to "none", anOffsetOutOfRangeException
will be thrown.Note that, the seek offset won't change to the in-flight fetch request, it will take effect in next fetch request. So, the consumer might wait for
fetch.max.wait.ms
before starting to fetch the records from desired offset.- Specified by:
seek
in interfaceConsumer<K,
V> - Parameters:
partition
- the TopicPartition on which the seek will be performed.offset
- the next offset returned by poll().- Throws:
IllegalArgumentException
- if the provided offset is negativeIllegalStateException
- if the provided TopicPartition is not assigned to this consumer- See Also:
-
seek
Overrides the fetch offsets that the consumer will use on the nextpoll(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. This method allows for setting the leaderEpoch along with the desired offset.- Specified by:
seek
in interfaceConsumer<K,
V> - Throws:
IllegalArgumentException
- if the provided offset is negativeIllegalStateException
- if the provided TopicPartition is not assigned to this consumer- See Also:
-
seekToBeginning
Seek to the first offset for each of the given partitions. This function evaluates lazily, seeking to the first offset in all partitions only whenpoll(Duration)
orposition(TopicPartition)
are called. If no partitions are provided, seek to the first offset for all of the currently assigned partitions.- Specified by:
seekToBeginning
in interfaceConsumer<K,
V> - Throws:
IllegalArgumentException
- ifpartitions
isnull
IllegalStateException
- if any of the provided partitions are not currently assigned to this consumer- See Also:
-
seekToEnd
Seek to the last offset for each of the given partitions. This function evaluates lazily, seeking to the final offset in all partitions only whenpoll(Duration)
orposition(TopicPartition)
are called. If no partitions are provided, seek to the final offset for all of the currently assigned partitions.If
isolation.level=read_committed
, the end offset will be the Last Stable Offset, i.e., the offset of the first message with an open transaction.- Specified by:
seekToEnd
in interfaceConsumer<K,
V> - Throws:
IllegalArgumentException
- ifpartitions
isnull
IllegalStateException
- if any of the provided partitions are not currently assigned to this consumer- See Also:
-
position
Get the offset of the next record that will be fetched (if a record with that offset exists). This method may issue a remote call to the server if there is no current position for the given partition.This call will block until either the position could be determined or an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout specified by
default.api.timeout.ms
expires (in which case aTimeoutException
is thrown to the caller).- Specified by:
position
in interfaceConsumer<K,
V> - Parameters:
partition
- The partition to get the position for- Returns:
- The current position of the consumer (that is, the offset of the next record to be fetched)
- Throws:
IllegalStateException
- if the provided TopicPartition is not assigned to this consumerInvalidOffsetException
- if no offset is currently defined for the partitionWakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsUnsupportedVersionException
- if the consumer attempts to fetch stable offsets when the broker doesn't support this featureKafkaException
- for any other unrecoverable errorsTimeoutException
- if the position cannot be determined before the timeout specified bydefault.api.timeout.ms
expires- See Also:
-
position
Get the offset of the next record that will be fetched (if a record with that offset exists). This method may issue a remote call to the server if there is no current position for the given partition.This call will block until the position can be determined, an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout expires.
- Specified by:
position
in interfaceConsumer<K,
V> - Parameters:
partition
- The partition to get the position fortimeout
- The maximum amount of time to await determination of the current position- Returns:
- The current position of the consumer (that is, the offset of the next record to be fetched)
- Throws:
IllegalStateException
- if the provided TopicPartition is not assigned to this consumerInvalidOffsetException
- if no offset is currently defined for the partitionWakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledTimeoutException
- if the position cannot be determined before the passed timeout expiresAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errors- See Also:
-
committed
Deprecated.since 2.4 Usecommitted(Set)
insteadGet 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 will do a remote call to get the latest committed offset from the server, and will block until the committed offset is gotten successfully, an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout specified by
default.api.timeout.ms
expires (in which case aTimeoutException
is thrown to the caller).- Specified by:
committed
in interfaceConsumer<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
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errorsTimeoutException
- if the committed offset cannot be found before the timeout specified bydefault.api.timeout.ms
expires.- See Also:
-
committed
Deprecated.since 2.4 Usecommitted(Set, Duration)
insteadGet 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 will block until the position can be determined, an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout expires.
- Specified by:
committed
in interfaceConsumer<K,
V> - Parameters:
partition
- The partition to checktimeout
- The maximum amount of time to await the current committed offset- Returns:
- The last committed offset and metadata or null if there was no prior commit
- Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errorsTimeoutException
- if the committed offset cannot be found before expiration of the timeout- See Also:
-
committed
Get the last committed offsets for the given partitions (whether the commit happened by this process or another). The returned offsets will be used as the position for the consumer in the event of a failure.If any of the partitions requested do not exist, an exception would be thrown.
This call will do a remote call to get the latest committed offsets from the server, and will block until the committed offsets are gotten successfully, an unrecoverable error is encountered (in which case it is thrown to the caller), or the timeout specified by
default.api.timeout.ms
expires (in which case aTimeoutException
is thrown to the caller).- Specified by:
committed
in interfaceConsumer<K,
V> - Parameters:
partitions
- The partitions to check- Returns:
- The latest committed offsets for the given partitions;
null
will be returned for the partition if there is no such message. - Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsUnsupportedVersionException
- if the consumer attempts to fetch stable offsets when the broker doesn't support this featureKafkaException
- for any other unrecoverable errorsTimeoutException
- if the committed offset cannot be found before the timeout specified bydefault.api.timeout.ms
expires.- See Also:
-
committed
public Map<TopicPartition,OffsetAndMetadata> committed(Set<TopicPartition> partitions, Duration timeout) Get the last committed offsets for the given partitions (whether the commit happened by this process or another). The returned offsets will be used as the position for the consumer in the event of a failure.If any of the partitions requested do not exist, an exception would be thrown.
This call will block to do a remote call to get the latest committed offsets from the server.
- Specified by:
committed
in interfaceConsumer<K,
V> - Parameters:
partitions
- The partitions to checktimeout
- The maximum amount of time to await the latest committed offsets- Returns:
- The latest committed offsets for the given partitions;
null
will be returned for the partition if there is no such message. - Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic or to the configured groupId. See the exception for more detailsKafkaException
- for any other unrecoverable errorsTimeoutException
- if the committed offset cannot be found before expiration of the timeout- See Also:
-
clientInstanceId
Determines the client's unique client instance ID used for telemetry. This ID is unique to this specific client instance and will not change after it is initially generated. The ID is useful for correlating client operations with telemetry sent to the broker and to its eventual monitoring destinations.If telemetry is enabled, this will first require a connection to the cluster to generate the unique client instance ID. This method waits up to
timeout
for the consumer client to complete the request.Client telemetry is controlled by the
ConsumerConfig.ENABLE_METRICS_PUSH_CONFIG
configuration option.- Specified by:
clientInstanceId
in interfaceConsumer<K,
V> - Parameters:
timeout
- The maximum time to wait for consumer client to determine its client instance ID. The value must be non-negative. Specifying a timeout of zero means do not wait for the initial request to complete if it hasn't already.- Returns:
- The client's assigned instance id used for metrics collection.
- Throws:
InterruptException
- If the thread is interrupted while blocked.KafkaException
- If an unexpected error occurs while trying to determine the client instance ID, though this error does not necessarily imply the consumer client is otherwise unusable.IllegalArgumentException
- If thetimeout
is negative.IllegalStateException
- If telemetry is not enabled ie, config `enable.metrics.push
` is set to `false
`.
-
metrics
Get the metrics kept by the consumer -
partitionsFor
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 interfaceConsumer<K,
V> - Parameters:
topic
- The topic to get partition metadata for- Returns:
- The list of partitions, which will be empty when the given topic is not found
- Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the specified topic. See the exception for more detailsKafkaException
- for any other unrecoverable errorsTimeoutException
- if the offset metadata could not be fetched before the amount of time allocated bydefault.api.timeout.ms
expires.- See Also:
-
partitionsFor
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 interfaceConsumer<K,
V> - Parameters:
topic
- The topic to get partition metadata fortimeout
- The maximum of time to await topic metadata- Returns:
- The list of partitions, which will be empty when the given topic is not found
- Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledAuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the specified topic. See the exception for more detailsTimeoutException
- if topic metadata cannot be fetched before expiration of the passed timeoutKafkaException
- for any other unrecoverable errors- See Also:
-
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 interfaceConsumer<K,
V> - Returns:
- The map of topics and its partitions
- Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledKafkaException
- for any other unrecoverable errorsTimeoutException
- if the offset metadata could not be fetched before the amount of time allocated bydefault.api.timeout.ms
expires.- See Also:
-
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 interfaceConsumer<K,
V> - Parameters:
timeout
- The maximum time this operation will block to fetch topic metadata- Returns:
- The map of topics and its partitions
- Throws:
WakeupException
- ifwakeup()
is called before or while this function is calledInterruptException
- if the calling thread is interrupted before or while this function is calledTimeoutException
- if the topic metadata could not be fetched before expiration of the passed timeoutKafkaException
- for any other unrecoverable errors- See Also:
-
pause
Suspend fetching from the requested partitions. Future calls topoll(Duration)
will not return any records from these partitions until they have been resumed usingresume(Collection)
. Note that this method does not affect partition subscription. In particular, it does not cause a group rebalance when automatic assignment is used. Note: Rebalance will not preserve the pause/resume state.- Specified by:
pause
in interfaceConsumer<K,
V> - Parameters:
partitions
- The partitions which should be paused- Throws:
IllegalStateException
- if any of the provided partitions are not currently assigned to this consumer- See Also:
-
resume
Resume specified partitions which have been paused withpause(Collection)
. New calls topoll(Duration)
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 interfaceConsumer<K,
V> - Parameters:
partitions
- The partitions which should be resumed- Throws:
IllegalStateException
- if any of the provided partitions are not currently assigned to this consumer- See Also:
-
paused
Get the set of partitions that were previously paused by a call topause(Collection)
. -
offsetsForTimes
public Map<TopicPartition,OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch) Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition. This is a blocking call. The consumer does not have to be assigned the partitions. If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, null will be returned for that partition.- Specified by:
offsetsForTimes
in interfaceConsumer<K,
V> - Parameters:
timestampsToSearch
- the mapping from partition to the timestamp to look up.- Returns:
- a mapping from partition to the timestamp and offset of the first message with timestamp greater
than or equal to the target timestamp.
null
will be returned for the partition if there is no such message. - Throws:
AuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic(s). See the exception for more detailsIllegalArgumentException
- if the target timestamp is negativeTimeoutException
- if the offset metadata could not be fetched before the amount of time allocated bydefault.api.timeout.ms
expires.UnsupportedVersionException
- if the broker does not support looking up the offsets by timestamp- See Also:
-
offsetsForTimes
public Map<TopicPartition,OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch, Duration timeout) Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition. This is a blocking call. The consumer does not have to be assigned the partitions. If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, null will be returned for that partition.- Specified by:
offsetsForTimes
in interfaceConsumer<K,
V> - Parameters:
timestampsToSearch
- the mapping from partition to the timestamp to look up.timeout
- The maximum amount of time to await retrieval of the offsets- Returns:
- a mapping from partition to the timestamp and offset of the first message with timestamp greater
than or equal to the target timestamp.
null
will be returned for the partition if there is no such message. - Throws:
AuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic(s). See the exception for more detailsIllegalArgumentException
- if the target timestamp is negativeTimeoutException
- if the offset metadata could not be fetched before expiration of the passed timeoutUnsupportedVersionException
- if the broker does not support looking up the offsets by timestamp- See Also:
-
beginningOffsets
Get the first offset for the given partitions.This method does not change the current consumer position of the partitions.
- Specified by:
beginningOffsets
in interfaceConsumer<K,
V> - Parameters:
partitions
- the partitions to get the earliest offsets.- Returns:
- The earliest available offsets for the given partitions
- Throws:
AuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic(s). See the exception for more detailsTimeoutException
- if the offset metadata could not be fetched before expiration of the configureddefault.api.timeout.ms
- See Also:
-
beginningOffsets
public Map<TopicPartition,Long> beginningOffsets(Collection<TopicPartition> partitions, Duration timeout) Get the first offset for the given partitions.This method does not change the current consumer position of the partitions.
- Specified by:
beginningOffsets
in interfaceConsumer<K,
V> - Parameters:
partitions
- the partitions to get the earliest offsetstimeout
- The maximum amount of time to await retrieval of the beginning offsets- Returns:
- The earliest available offsets for the given partitions
- Throws:
AuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic(s). See the exception for more detailsTimeoutException
- if the offset metadata could not be fetched before expiration of the passed timeout- See Also:
-
endOffsets
Get the end offsets for the given partitions. In the defaultread_uncommitted
isolation level, the end offset is the high watermark (that is, the offset of the last successfully replicated message plus one). Forread_committed
consumers, the end offset is the last stable offset (LSO), which is the minimum of the high watermark and the smallest offset of any open transaction. Finally, if the partition has never been written to, the end offset is 0.This method does not change the current consumer position of the partitions.
- Specified by:
endOffsets
in interfaceConsumer<K,
V> - Parameters:
partitions
- the partitions to get the end offsets.- Returns:
- The end offsets for the given partitions.
- Throws:
AuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic(s). See the exception for more detailsTimeoutException
- if the offset metadata could not be fetched before the amount of time allocated bydefault.api.timeout.ms
expires- See Also:
-
endOffsets
Get the end offsets for the given partitions. In the defaultread_uncommitted
isolation level, the end offset is the high watermark (that is, the offset of the last successfully replicated message plus one). Forread_committed
consumers, the end offset is the last stable offset (LSO), which is the minimum of the high watermark and the smallest offset of any open transaction. Finally, if the partition has never been written to, the end offset is 0.This method does not change the current consumer position of the partitions.
- Specified by:
endOffsets
in interfaceConsumer<K,
V> - Parameters:
partitions
- the partitions to get the end offsets.timeout
- The maximum amount of time to await retrieval of the end offsets- Returns:
- The end offsets for the given partitions.
- Throws:
AuthenticationException
- if authentication fails. See the exception for more detailsAuthorizationException
- if not authorized to the topic(s). See the exception for more detailsTimeoutException
- if the offsets could not be fetched before expiration of the passed timeout- See Also:
-
currentLag
Get the consumer's current lag on the partition. Returns an "empty"OptionalLong
if the lag is not known, for example if there is no position yet, or if the end offset is not known yet.This method uses locally cached metadata. If the log end offset is not known yet, it triggers a request to fetch the log end offset, but returns immediately.
- Specified by:
currentLag
in interfaceConsumer<K,
V> - Parameters:
topicPartition
- The partition to get the lag for.- Returns:
- This
Consumer
instance's current lag for the given partition. - Throws:
IllegalStateException
- if thetopicPartition
is not assigned- See Also:
-
groupMetadata
Return the current group metadata associated with this consumer.- Specified by:
groupMetadata
in interfaceConsumer<K,
V> - Returns:
- consumer group metadata
- Throws:
InvalidGroupIdException
- if consumer does not have a group- See Also:
-
enforceRebalance
Alert the consumer to trigger a new rebalance by rejoining the group. This is a nonblocking call that forces the consumer to trigger a new rebalance on the nextpoll(Duration)
call. Note that this API does not itself initiate the rebalance, so you must still callpoll(Duration)
. If a rebalance is already in progress this call will be a no-op. If you wish to force an additional rebalance you must complete the current one by calling poll before retrying this API.You do not need to call this during normal processing, as the consumer group will manage itself automatically and rebalance when necessary. However there may be situations where the application wishes to trigger a rebalance that would otherwise not occur. For example, if some condition external and invisible to the Consumer and its group changes in a way that would affect the userdata encoded in the
Subscription
, the Consumer will not be notified and no rebalance will occur. This API can be used to force the group to rebalance so that the assignor can perform a partition reassignment based on the latest userdata. If your assignor does not use this userdata, or you do not use a customConsumerPartitionAssignor
, you should not use this API.- Specified by:
enforceRebalance
in interfaceConsumer<K,
V> - Parameters:
reason
- The reason why the new rebalance is needed.- Throws:
IllegalStateException
- if the consumer does not use group subscription- See Also:
-
enforceRebalance
public void enforceRebalance()- Specified by:
enforceRebalance
in interfaceConsumer<K,
V> - See Also:
-
close
public void close()Close the consumer, waiting for up to the default timeout of 30 seconds for any needed cleanup. If auto-commit is enabled, this will commit the current offsets if possible within the default timeout. Seeclose(Duration)
for details. Note thatwakeup()
cannot be used to interrupt close.- Specified by:
close
in interfaceAutoCloseable
- Specified by:
close
in interfaceCloseable
- Specified by:
close
in interfaceConsumer<K,
V> - Throws:
InterruptException
- if the calling thread is interrupted before or while this function is calledKafkaException
- for any other error during close- See Also:
-
close
Tries to close the consumer cleanly within the specified timeout. This method waits up totimeout
for the consumer to complete pending commits and leave the group. If auto-commit is enabled, this will commit the current offsets if possible within the timeout. If the consumer is unable to complete offset commits and gracefully leave the group before the timeout expires, the consumer is force closed. Note thatwakeup()
cannot be used to interrupt close.- Specified by:
close
in interfaceConsumer<K,
V> - Parameters:
timeout
- The maximum time to wait for consumer to close gracefully. The value must be non-negative. Specifying a timeout of zero means do not wait for pending requests to complete.- Throws:
IllegalArgumentException
- If thetimeout
is negative.InterruptException
- If the thread is interrupted before or while this function is calledKafkaException
- for any other error during close- See Also:
-
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 throwWakeupException
. If no thread is blocking in a method which can throwWakeupException
, the next call to such a method will raise it instead.
-
committed(Set)
instead