You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gw...@apache.org on 2016/05/18 04:54:07 UTC
[25/30] kafka-site git commit: Doc refresh for 0.10.0.0-rc6 and
getting the javadoc for Java clients, not the old Java-API in core
http://git-wip-us.apache.org/repos/asf/kafka-site/blob/b91f7909/0100/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
----------------------------------------------------------------------
diff --git a/0100/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html b/0100/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
new file mode 100644
index 0000000..214b72e
--- /dev/null
+++ b/0100/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
@@ -0,0 +1,1463 @@
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
+<!-- NewPage -->
+<html lang="en">
+<head>
+<!-- Generated by javadoc (version 1.7.0_79) on Tue May 17 21:36:37 PDT 2016 -->
+<title>KafkaConsumer (clients 0.10.0.0 API)</title>
+<meta name="date" content="2016-05-17">
+<link rel="stylesheet" type="text/css" href="../../../../../stylesheet.css" title="Style">
+</head>
+<body>
+<script type="text/javascript"><!--
+ if (location.href.indexOf('is-external=true') == -1) {
+ parent.document.title="KafkaConsumer (clients 0.10.0.0 API)";
+ }
+//-->
+</script>
+<noscript>
+<div>JavaScript is disabled on your browser.</div>
+</noscript>
+<!-- ========= START OF TOP NAVBAR ======= -->
+<div class="topNav"><a name="navbar_top">
+<!-- -->
+</a><a href="#skip-navbar_top" title="Skip navigation links"></a><a name="navbar_top_firstrow">
+<!-- -->
+</a>
+<ul class="navList" title="Navigation">
+<li><a href="../../../../../overview-summary.html">Overview</a></li>
+<li><a href="package-summary.html">Package</a></li>
+<li class="navBarCell1Rev">Class</li>
+<li><a href="package-tree.html">Tree</a></li>
+<li><a href="../../../../../deprecated-list.html">Deprecated</a></li>
+<li><a href="../../../../../index-all.html">Index</a></li>
+<li><a href="../../../../../help-doc.html">Help</a></li>
+</ul>
+</div>
+<div class="subNav">
+<ul class="navList">
+<li><a href="../../../../../org/apache/kafka/clients/consumer/InvalidOffsetException.html" title="class in org.apache.kafka.clients.consumer"><span class="strong">Prev Class</span></a></li>
+<li><a href="../../../../../org/apache/kafka/clients/consumer/MockConsumer.html" title="class in org.apache.kafka.clients.consumer"><span class="strong">Next Class</span></a></li>
+</ul>
+<ul class="navList">
+<li><a href="../../../../../index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html" target="_top">Frames</a></li>
+<li><a href="KafkaConsumer.html" target="_top">No Frames</a></li>
+</ul>
+<ul class="navList" id="allclasses_navbar_top">
+<li><a href="../../../../../allclasses-noframe.html">All Classes</a></li>
+</ul>
+<div>
+<script type="text/javascript"><!--
+ allClassesLink = document.getElementById("allclasses_navbar_top");
+ if(window==top) {
+ allClassesLink.style.display = "block";
+ }
+ else {
+ allClassesLink.style.display = "none";
+ }
+ //-->
+</script>
+</div>
+<div>
+<ul class="subNavList">
+<li>Summary: </li>
+<li>Nested | </li>
+<li>Field | </li>
+<li><a href="#constructor_summary">Constr</a> | </li>
+<li><a href="#method_summary">Method</a></li>
+</ul>
+<ul class="subNavList">
+<li>Detail: </li>
+<li>Field | </li>
+<li><a href="#constructor_detail">Constr</a> | </li>
+<li><a href="#method_detail">Method</a></li>
+</ul>
+</div>
+<a name="skip-navbar_top">
+<!-- -->
+</a></div>
+<!-- ========= END OF TOP NAVBAR ========= -->
+<!-- ======== START OF CLASS DATA ======== -->
+<div class="header">
+<div class="subTitle">org.apache.kafka.clients.consumer</div>
+<h2 title="Class KafkaConsumer" class="title">Class KafkaConsumer<K,V></h2>
+</div>
+<div class="contentContainer">
+<ul class="inheritance">
+<li>java.lang.Object</li>
+<li>
+<ul class="inheritance">
+<li>org.apache.kafka.clients.consumer.KafkaConsumer<K,V></li>
+</ul>
+</li>
+</ul>
+<div class="description">
+<ul class="blockList">
+<li class="blockList">
+<dl>
+<dt>All Implemented Interfaces:</dt>
+<dd>java.io.Closeable, java.lang.AutoCloseable, <a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><K,V></dd>
+</dl>
+<hr>
+<br>
+<pre>public class <span class="strong">KafkaConsumer<K,V></span>
+extends java.lang.Object
+implements <a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><K,V></pre>
+<div class="block">A Kafka client that consumes records from a Kafka cluster.
+ <p>
+ 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).
+ <p>
+ 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 <a href="#multithreaded">Multi-threaded Processing</a> for more details.
+
+ <h3>Offsets and Consumer Position</h3>
+ 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 the record with offset 5. There
+ are actually two notions of position relevant to the user of the consumer.
+ <p>
+ The <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#position(org.apache.kafka.common.TopicPartition)"><code>position</code></a> 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 <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a> and receives messages.
+ <p>
+ The <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync()"><code>committed position</code></a> 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
+ <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync()"><code>commitSync</code></a>, which will block until the offsets have been successfully committed
+ or fatal error has happened during the commit process, or <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitAsync(org.apache.kafka.clients.consumer.OffsetCommitCallback)"><code>commitAsync</code></a> which is non-blocking
+ and will trigger <a href="../../../../../org/apache/kafka/clients/consumer/OffsetCommitCallback.html" title="interface in org.apache.kafka.clients.consumer"><code>OffsetCommitCallback</code></a> upon either successfully committed or fatally failed.
+ <p>
+ This distinction gives the consumer control over when a record is considered consumed. It is discussed in further
+ detail below.
+
+ <h3>Consumer Groups and Topic Subscriptions</h3>
+
+ Kafka uses the concept of <i>consumer groups</i> 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, as is more likely, they can be
+ distributed over many machines to provide scalability and fault tolerance for processing.
+ <p>
+ 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 one of the <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe</code></a>
+ 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.
+ <p>
+ 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 <i>rebalancing</i> the group and is discussed in more
+ detail <a href="#failuredetection">below</a>. Note that the same process is also used when new partitions are added
+ to one of the subscribed topics: the group automatically detects the new partitions and rebalances the group so
+ that every new partition is assigned to one of the members.
+ <p>
+ 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).
+ <p>
+ 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.
+ <p>
+ In addition, when group reassignment happens automatically, consumers can be notified through <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer"><code>ConsumerRebalanceListener</code></a>,
+ 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 <a href="#rebalancecallback">Storing Offsets Outside Kafka</a> for more details
+ <p>
+ It is also possible for the consumer to <a href="#manualassignment">manually assign</a> specific partitions
+ (similar to the older "simple" consumer) using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a>. In this case, dynamic partition
+ assignment and consumer group coordination will be disabled.
+
+ <h3><a name="failuredetection">Detecting Consumer Failures</a></h3>
+
+ After subscribing to a set of topics, the consumer will automatically join the group when <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a> 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 poll API sends periodic heartbeats to the server; when you stop calling poll (perhaps because an exception was thrown),
+ then no heartbeats will be sent. If a period of the configured <i>session timeout</i> elapses before the server
+ has received a heartbeat, then the consumer will be kicked out of the group and its partitions will be reassigned.
+ This is designed to prevent situations where the consumer has failed, yet continues to hold onto the partitions
+ it was assigned (thus preventing active consumers in the group from taking them). To stay in the group, you
+ have to prove you are still alive by calling poll.
+ <p>
+ The implication of this design is that message processing time in the poll loop must be bounded so that
+ heartbeats can be sent before expiration of the session timeout. What typically happens when processing time
+ exceeds the session timeout is that the consumer won't be able to commit offsets for any of the processed records.
+ For example, this is indicated by a <a href="../../../../../org/apache/kafka/clients/consumer/CommitFailedException.html" title="class in org.apache.kafka.clients.consumer"><code>CommitFailedException</code></a> thrown from <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync()"><code>commitSync()</code></a>. This
+ guarantees that only active members of the group are allowed to commit offsets. If the consumer
+ has been kicked out of the group, then its partitions will have been assigned to another member, which will be
+ committing its own offsets as it handles new records. This gives offset commits an isolation guarantee.
+ <p>
+ The consumer provides two configuration settings to control this behavior:
+ <ol>
+ <li><code>session.timeout.ms</code>: By increasing the session timeout, you can give the consumer more
+ time to handle a batch of records returned from <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a>. The only drawback is that it
+ will take longer for the server to detect hard consumer failures, which can cause a delay before
+ a rebalance can be completed. However, clean shutdown with <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#close()"><code>close()</code></a> is not impacted since
+ the consumer will send an explicit message to the server to leave the group and cause an immediate
+ rebalance.</li>
+ <li><code>max.poll.records</code>: Processing time in the poll loop is typically proportional to the number
+ of records processed, so it's natural to want to set a limit on the number of records handled at once.
+ This setting provides that. By default, there is essentially no limit.</li>
+ </ol>
+ <p>
+ For use cases where message processing time varies unpredictably, neither of these options may be viable.
+ The recommended way to handle these cases is to move message processing to another thread, which allows
+ the consumer to continue sending heartbeats 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 generally need to <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)"><code>pause(Collection)</code></a>
+ the partition so that no new records are received from poll until after thread has finished handling those
+ previously returned.
+
+ <h3>Usage Examples</h3>
+ The consumer APIs offer flexibility to cover a variety of consumption use cases. Here are some examples to
+ demonstrate how to use them.
+
+ <h4>Automatic Offset Committing</h4>
+ This example demonstrates a simple usage of Kafka's consumer api that relying on automatic offset committing.
+ <p>
+ <pre>
+ 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());
+ }
+ </pre>
+
+ Setting <code>enable.auto.commit</code> means that offsets are committed automatically with a frequency controlled by
+ the config <code>auto.commit.interval.ms</code>.
+ <p>
+ The connection to the cluster is bootstrapped by specifying a list of one or more brokers to contact using the
+ configuration <code>bootstrap.servers</code>. 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).
+ <p>
+ In this example the client is subscribing to the topics <i>foo</i> and <i>bar</i> as part of a group of consumers
+ called <i>test</i> as described above.
+ <p>
+ The broker will automatically detect failed processes in the <i>test</i> group by using a heartbeat mechanism. The
+ consumer will automatically ping the cluster periodically, which lets the cluster know that it is alive. Note that
+ the consumer is single-threaded, so periodic heartbeats can only be sent when <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a> is called. 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 by failing to call <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a> for a period of time longer than <code>session.timeout.ms</code>
+ then it will be considered dead and its partitions will be assigned to another process.
+ <p>
+ 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.
+
+ <h4>Manual Offset Control</h4>
+
+ 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.
+ <p>
+ <pre>
+ 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();
+ }
+ }
+ </pre>
+
+ The above example uses <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync()"><code>commitSync</code></a> 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.
+ <p>
+ <pre>
+ 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();
+ }
+ </pre>
+
+ <b>Note: The committed offset should always be the offset of the next message that your application will read.</b>
+ Thus, when calling <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync(java.util.Map)"><code>commitSync(offsets)</code></a> you should add one to the offset of the last message processed.
+
+ <h4><a name="manualassignment">Manual Partition Assignment</a></h4>
+
+ 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:
+ <p>
+ <ul>
+ <li>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.
+ <li>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.
+ </ul>
+ <p>
+ To use this mode, instead of subscribing to the topic using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)"><code>subscribe</code></a>, you just call
+ <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a> with the full list of partitions that you want to consume.
+
+ <pre>
+ String topic = "foo";
+ TopicPartition partition0 = new TopicPartition(topic, 0);
+ TopicPartition partition1 = new TopicPartition(topic, 1);
+ consumer.assign(Arrays.asList(partition0, partition1));
+ </pre>
+
+ Once assigned, you can call <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll</code></a> 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 <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign</code></a>. 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.
+ <p>
+ Note that it isn't possible to mix manual partition assignment (i.e. using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign</code></a>)
+ with dynamic partition assignment through topic subscription (i.e. using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)"><code>subscribe</code></a>).
+
+ <h4><a name="rebalancecallback">Storing Offsets Outside Kafka</h4>
+
+ 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.
+ <p>
+ Here are a couple of examples of this type of usage:
+ <ul>
+ <li>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.
+ <li>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.
+ </ul>
+ <p>
+ Each record comes with its own offset, so to manage your own offset you just need to do the following:
+
+ <ul>
+ <li>Configure <code>enable.auto.commit=false</code>
+ <li>Use the offset provided with each <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRecord.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerRecord</code></a> to save your position.
+ <li>On restart restore the position of the consumer using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seek(org.apache.kafka.common.TopicPartition,%20long)"><code>seek(TopicPartition, long)</code></a>.
+ </ul>
+
+ <p>
+ 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
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer"><code>ConsumerRebalanceListener</code></a> instance in the call to <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Collection, ConsumerRebalanceListener)</code></a>
+ and <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.regex.Pattern,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Pattern, ConsumerRebalanceListener)</code></a>.
+ For example, when partitions are taken from a consumer the consumer will want to commit its offset for those partitions by
+ implementing <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html#onPartitionsRevoked(java.util.Collection)"><code>ConsumerRebalanceListener.onPartitionsRevoked(Collection)</code></a>. 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 <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html#onPartitionsAssigned(java.util.Collection)"><code>ConsumerRebalanceListener.onPartitionsAssigned(Collection)</code></a>.
+ <p>
+ Another common use for <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer"><code>ConsumerRebalanceListener</code></a> is to flush any caches the application maintains for
+ partitions that are moved elsewhere.
+
+ <h4>Controlling The Consumer's Position</h4>
+
+ 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.
+ <p>
+ There are several instances where manually controlling the consumer's position can be useful.
+ <p>
+ 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.
+ <p>
+ 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).
+ <p>
+ Kafka allows specifying the position using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seek(org.apache.kafka.common.TopicPartition,%20long)"><code>seek(TopicPartition, long)</code></a> to specify the new position. Special
+ methods for seeking to the earliest and latest offset the server maintains are also available (
+ <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seekToBeginning(java.util.Collection)"><code>seekToBeginning(Collection)</code></a> and <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seekToEnd(java.util.Collection)"><code>seekToEnd(Collection)</code></a> respectively).
+
+ <h4>Consumption Flow Control</h4>
+
+ 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.
+
+ <p>
+ 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 bootstraping 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.
+
+ <p>
+ Kafka supports dynamic controlling of consumption flows by using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)"><code>pause(Collection)</code></a> and <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#resume(java.util.Collection)"><code>resume(Collection)</code></a>
+ to pause the consumption on the specified assigned partitions and resume the consumption
+ on the specified paused partitions respectively in the future <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a> calls.
+
+ <h3><a name="multithreaded">Multi-threaded Processing</a></h3>
+
+ 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 <code>ConcurrentModificationException</code>.
+
+ <p>
+ The only exception to this rule is <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#wakeup()"><code>wakeup()</code></a>, which can safely be used from an external thread to
+ interrupt an active operation. In this case, a <a href="../../../../../org/apache/kafka/common/errors/WakeupException.html" title="class in org.apache.kafka.common.errors"><code>WakeupException</code></a> 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:
+
+ <pre>
+ 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();
+ }
+ }
+ </pre>
+
+ Then in a separate thread, the consumer can be shutdown by setting the closed flag and waking up the consumer.
+
+ <p>
+ <pre>
+ closed.set(true);
+ consumer.wakeup();
+ </pre>
+
+ <p>
+ We have intentionally avoided implementing a particular threading model for processing. This leaves several
+ options for implementing multi-threaded processing of records.
+
+
+ <h4>1. One Consumer Per Thread</h4>
+
+ A simple option is to give each thread its own consumer instance. Here are the pros and cons of this approach:
+ <ul>
+ <li><b>PRO</b>: It is the easiest to implement
+ <li><b>PRO</b>: It is often the fastest as no inter-thread co-ordination is needed
+ <li><b>PRO</b>: 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).
+ <li><b>CON</b>: 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.
+ <li><b>CON</b>: 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.
+ <li><b>CON</b>: The number of total threads across all processes will be limited by the total number of partitions.
+ </ul>
+
+ <h4>2. Decouple Consumption and Processing</h4>
+
+ Another alternative is to have one or more consumer threads that do all data consumption and hands off
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRecords.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerRecords</code></a> 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:
+ <ul>
+ <li><b>PRO</b>: 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.
+ <li><b>CON</b>: 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.
+ <li><b>CON</b>: Manually committing the position becomes harder as it requires that all threads co-ordinate to ensure
+ that processing is complete for that partition.
+ </ul>
+
+ 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.</div>
+</li>
+</ul>
+</div>
+<div class="summary">
+<ul class="blockList">
+<li class="blockList">
+<!-- ======== CONSTRUCTOR SUMMARY ======== -->
+<ul class="blockList">
+<li class="blockList"><a name="constructor_summary">
+<!-- -->
+</a>
+<h3>Constructor Summary</h3>
+<table class="overviewSummary" border="0" cellpadding="3" cellspacing="0" summary="Constructor Summary table, listing constructors, and an explanation">
+<caption><span>Constructors</span><span class="tabEnd"> </span></caption>
+<tr>
+<th class="colOne" scope="col">Constructor and Description</th>
+</tr>
+<tr class="altColor">
+<td class="colOne"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#KafkaConsumer(java.util.Map)">KafkaConsumer</a></strong>(java.util.Map<java.lang.String,java.lang.Object> configs)</code>
+<div class="block">A consumer is instantiated by providing a set of key-value pairs as configuration.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colOne"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#KafkaConsumer(java.util.Map,%20org.apache.kafka.common.serialization.Deserializer,%20org.apache.kafka.common.serialization.Deserializer)">KafkaConsumer</a></strong>(java.util.Map<java.lang.String,java.lang.Object> configs,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>> keyDeserializer,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>> valueDeserializer)</code>
+<div class="block">A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colOne"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#KafkaConsumer(java.util.Properties)">KafkaConsumer</a></strong>(java.util.Properties properties)</code>
+<div class="block">A consumer is instantiated by providing a <code>Properties</code> object as configuration.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colOne"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#KafkaConsumer(java.util.Properties,%20org.apache.kafka.common.serialization.Deserializer,%20org.apache.kafka.common.serialization.Deserializer)">KafkaConsumer</a></strong>(java.util.Properties properties,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>> keyDeserializer,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>> valueDeserializer)</code>
+<div class="block">A consumer is instantiated by providing a <code>Properties</code> object as configuration, and a
+ key and a value <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>.</div>
+</td>
+</tr>
+</table>
+</li>
+</ul>
+<!-- ========== METHOD SUMMARY =========== -->
+<ul class="blockList">
+<li class="blockList"><a name="method_summary">
+<!-- -->
+</a>
+<h3>Method Summary</h3>
+<table class="overviewSummary" border="0" cellpadding="3" cellspacing="0" summary="Method Summary table, listing methods, and an explanation">
+<caption><span>Methods</span><span class="tabEnd"> </span></caption>
+<tr>
+<th class="colFirst" scope="col">Modifier and Type</th>
+<th class="colLast" scope="col">Method and Description</th>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)">assign</a></strong>(java.util.Collection<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> partitions)</code>
+<div class="block">Manually assign a list of partition to this consumer.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>java.util.Set<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assignment()">assignment</a></strong>()</code>
+<div class="block">Get the set of partitions currently assigned to this consumer.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#close()">close</a></strong>()</code>
+<div class="block">Close the consumer, waiting indefinitely for any needed cleanup.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitAsync()">commitAsync</a></strong>()</code>
+<div class="block">Commit offsets returned on the last <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll()</code></a> for all the subscribed list of topics and partition.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitAsync(java.util.Map,%20org.apache.kafka.clients.consumer.OffsetCommitCallback)">commitAsync</a></strong>(java.util.Map<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>,<a href="../../../../../org/apache/kafka/clients/consumer/OffsetAndMetadata.html" title="class in org.apache.kafka.clients.consumer">OffsetAndMetadata</a>> offsets,
+ <a href="../../../../../org/apache/kafka/clients/consumer/OffsetCommitCallback.html" title="interface in org.apache.kafka.clients.consumer">OffsetCommitCallback</a> callback)</code>
+<div class="block">Commit the specified offsets for the specified list of topics and partitions to Kafka.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitAsync(org.apache.kafka.clients.consumer.OffsetCommitCallback)">commitAsync</a></strong>(<a href="../../../../../org/apache/kafka/clients/consumer/OffsetCommitCallback.html" title="interface in org.apache.kafka.clients.consumer">OffsetCommitCallback</a> callback)</code>
+<div class="block">Commit offsets returned on the last <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll()</code></a> for the subscribed list of topics and partitions.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync()">commitSync</a></strong>()</code>
+<div class="block">Commit offsets returned on the last <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll()</code></a> for all the subscribed list of topics and partitions.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync(java.util.Map)">commitSync</a></strong>(java.util.Map<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>,<a href="../../../../../org/apache/kafka/clients/consumer/OffsetAndMetadata.html" title="class in org.apache.kafka.clients.consumer">OffsetAndMetadata</a>> offsets)</code>
+<div class="block">Commit the specified offsets for the specified list of topics and partitions.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code><a href="../../../../../org/apache/kafka/clients/consumer/OffsetAndMetadata.html" title="class in org.apache.kafka.clients.consumer">OffsetAndMetadata</a></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#committed(org.apache.kafka.common.TopicPartition)">committed</a></strong>(<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a> partition)</code>
+<div class="block">Get the last committed offset for the given partition (whether the commit happened by this process or
+ another).</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>java.util.Map<java.lang.String,java.util.List<<a href="../../../../../org/apache/kafka/common/PartitionInfo.html" title="class in org.apache.kafka.common">PartitionInfo</a>>></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#listTopics()">listTopics</a></strong>()</code>
+<div class="block">Get metadata about partitions for all topics that the user is authorized to view.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>java.util.Map<<a href="../../../../../org/apache/kafka/common/MetricName.html" title="class in org.apache.kafka.common">MetricName</a>,? extends <a href="../../../../../org/apache/kafka/common/Metric.html" title="interface in org.apache.kafka.common">Metric</a>></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#metrics()">metrics</a></strong>()</code>
+<div class="block">Get the metrics kept by the consumer</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>java.util.List<<a href="../../../../../org/apache/kafka/common/PartitionInfo.html" title="class in org.apache.kafka.common">PartitionInfo</a>></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#partitionsFor(java.lang.String)">partitionsFor</a></strong>(java.lang.String topic)</code>
+<div class="block">Get metadata about the partitions for a given topic.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)">pause</a></strong>(java.util.Collection<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> partitions)</code>
+<div class="block">Suspend fetching from the requested partitions.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>java.util.Set<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#paused()">paused</a></strong>()</code>
+<div class="block">Get the set of partitions that were previously paused by a call to <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)"><code>pause(Collection)</code></a>.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code><a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRecords.html" title="class in org.apache.kafka.clients.consumer">ConsumerRecords</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)">poll</a></strong>(long timeout)</code>
+<div class="block">Fetch data for the topics or partitions specified using one of the subscribe/assign APIs.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>long</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#position(org.apache.kafka.common.TopicPartition)">position</a></strong>(<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a> partition)</code>
+<div class="block">Get the offset of the <i>next record</i> that will be fetched (if a record with that offset exists).</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#resume(java.util.Collection)">resume</a></strong>(java.util.Collection<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> partitions)</code>
+<div class="block">Resume specified partitions which have been paused with <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)"><code>pause(Collection)</code></a>.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seek(org.apache.kafka.common.TopicPartition,%20long)">seek</a></strong>(<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a> partition,
+ long offset)</code>
+<div class="block">Overrides the fetch offsets that the consumer will use on the next <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(timeout)</code></a>.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seekToBeginning(java.util.Collection)">seekToBeginning</a></strong>(java.util.Collection<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> partitions)</code>
+<div class="block">Seek to the first offset for each of the given partitions.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seekToEnd(java.util.Collection)">seekToEnd</a></strong>(java.util.Collection<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> partitions)</code>
+<div class="block">Seek to the last offset for each of the given partitions.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)">subscribe</a></strong>(java.util.Collection<java.lang.String> topics)</code>
+<div class="block">Subscribe to the given list of topics to get dynamically assigned partitions.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)">subscribe</a></strong>(java.util.Collection<java.lang.String> topics,
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer">ConsumerRebalanceListener</a> listener)</code>
+<div class="block">Subscribe to the given list of topics to get dynamically
+ assigned partitions.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.regex.Pattern,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)">subscribe</a></strong>(java.util.regex.Pattern pattern,
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer">ConsumerRebalanceListener</a> listener)</code>
+<div class="block">Subscribe to all topics matching specified pattern to get dynamically assigned partitions.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>java.util.Set<java.lang.String></code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscription()">subscription</a></strong>()</code>
+<div class="block">Get the current subscription.</div>
+</td>
+</tr>
+<tr class="altColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#unsubscribe()">unsubscribe</a></strong>()</code>
+<div class="block">Unsubscribe from topics currently subscribed with <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)"><code>subscribe(Collection)</code></a>.</div>
+</td>
+</tr>
+<tr class="rowColor">
+<td class="colFirst"><code>void</code></td>
+<td class="colLast"><code><strong><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#wakeup()">wakeup</a></strong>()</code>
+<div class="block">Wakeup the consumer.</div>
+</td>
+</tr>
+</table>
+<ul class="blockList">
+<li class="blockList"><a name="methods_inherited_from_class_java.lang.Object">
+<!-- -->
+</a>
+<h3>Methods inherited from class java.lang.Object</h3>
+<code>clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait</code></li>
+</ul>
+</li>
+</ul>
+</li>
+</ul>
+</div>
+<div class="details">
+<ul class="blockList">
+<li class="blockList">
+<!-- ========= CONSTRUCTOR DETAIL ======== -->
+<ul class="blockList">
+<li class="blockList"><a name="constructor_detail">
+<!-- -->
+</a>
+<h3>Constructor Detail</h3>
+<a name="KafkaConsumer(java.util.Map)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>KafkaConsumer</h4>
+<pre>public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)</pre>
+<div class="block">A consumer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings
+ are documented <a href="http://kafka.apache.org/documentation.html#consumerconfigs" >here</a>. 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).
+ <p>
+ Valid configuration strings are documented at <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerConfig.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerConfig</code></a></div>
+<dl><dt><span class="strong">Parameters:</span></dt><dd><code>configs</code> - The consumer configs</dd></dl>
+</li>
+</ul>
+<a name="KafkaConsumer(java.util.Map, org.apache.kafka.common.serialization.Deserializer, org.apache.kafka.common.serialization.Deserializer)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>KafkaConsumer</h4>
+<pre>public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>> keyDeserializer,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>> valueDeserializer)</pre>
+<div class="block">A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>.
+ <p>
+ Valid configuration strings are documented at <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerConfig.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerConfig</code></a></div>
+<dl><dt><span class="strong">Parameters:</span></dt><dd><code>configs</code> - The consumer configs</dd><dd><code>keyDeserializer</code> - The deserializer for key that implements <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>. The configure() method
+ won't be called in the consumer when the deserializer is passed in directly.</dd><dd><code>valueDeserializer</code> - The deserializer for value that implements <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>. The configure() method
+ won't be called in the consumer when the deserializer is passed in directly.</dd></dl>
+</li>
+</ul>
+<a name="KafkaConsumer(java.util.Properties)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>KafkaConsumer</h4>
+<pre>public KafkaConsumer(java.util.Properties properties)</pre>
+<div class="block">A consumer is instantiated by providing a <code>Properties</code> object as configuration. Valid
+ configuration strings are documented at <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerConfig.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerConfig</code></a> A consumer is instantiated by providing a
+ <code>Properties</code> object as configuration. Valid configuration strings are documented at
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerConfig.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerConfig</code></a></div>
+</li>
+</ul>
+<a name="KafkaConsumer(java.util.Properties, org.apache.kafka.common.serialization.Deserializer, org.apache.kafka.common.serialization.Deserializer)">
+<!-- -->
+</a>
+<ul class="blockListLast">
+<li class="blockList">
+<h4>KafkaConsumer</h4>
+<pre>public KafkaConsumer(java.util.Properties properties,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>> keyDeserializer,
+ <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization">Deserializer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>> valueDeserializer)</pre>
+<div class="block">A consumer is instantiated by providing a <code>Properties</code> object as configuration, and a
+ key and a value <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>.
+ <p>
+ Valid configuration strings are documented at <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerConfig.html" title="class in org.apache.kafka.clients.consumer"><code>ConsumerConfig</code></a></div>
+<dl><dt><span class="strong">Parameters:</span></dt><dd><code>properties</code> - The consumer configuration properties</dd><dd><code>keyDeserializer</code> - The deserializer for key that implements <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>. The configure() method
+ won't be called in the consumer when the deserializer is passed in directly.</dd><dd><code>valueDeserializer</code> - The deserializer for value that implements <a href="../../../../../org/apache/kafka/common/serialization/Deserializer.html" title="interface in org.apache.kafka.common.serialization"><code>Deserializer</code></a>. The configure() method
+ won't be called in the consumer when the deserializer is passed in directly.</dd></dl>
+</li>
+</ul>
+</li>
+</ul>
+<!-- ============ METHOD DETAIL ========== -->
+<ul class="blockList">
+<li class="blockList"><a name="method_detail">
+<!-- -->
+</a>
+<h3>Method Detail</h3>
+<a name="assignment()">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>assignment</h4>
+<pre>public java.util.Set<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> assignment()</pre>
+<div class="block">Get the set of partitions currently assigned to this consumer. If subscription happened by directly assigning
+ partitions using <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a> 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).</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#assignment()">assignment</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Returns:</span></dt><dd>The set of partitions currently assigned to this consumer</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assignment()"><code>assignment()</code></a></dd></dl>
+</li>
+</ul>
+<a name="subscription()">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>subscription</h4>
+<pre>public java.util.Set<java.lang.String> subscription()</pre>
+<div class="block">Get the current subscription. Will return the same topics used in the most recent call to
+ <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Collection, ConsumerRebalanceListener)</code></a>, or an empty set if no such call has been made.</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#subscription()">subscription</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Returns:</span></dt><dd>The set of topics currently subscribed to</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscription()"><code>subscription()</code></a></dd></dl>
+</li>
+</ul>
+<a name="subscribe(java.util.Collection, org.apache.kafka.clients.consumer.ConsumerRebalanceListener)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>subscribe</h4>
+<pre>public void subscribe(java.util.Collection<java.lang.String> topics,
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer">ConsumerRebalanceListener</a> listener)</pre>
+<div class="block">Subscribe to the given list of topics to get dynamically
+ assigned partitions. <b>Topic subscriptions are not incremental. This list will replace the current
+ assignment (if there is one).</b> Note that it is not possible to combine topic subscription with group management
+ with manual partition assignment through <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a>.
+
+ If the given list of topics is empty, it is treated the same as <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#unsubscribe()"><code>unsubscribe()</code></a>.
+
+ <p>
+ 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 -
+ <ul>
+ <li>Number of partitions change for any of the subscribed list of topics
+ <li>Topic is created or deleted
+ <li>An existing member of the consumer group dies
+ <li>A new member is added to an existing consumer group via the join API
+ </ul>
+ <p>
+ 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 <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer"><code>ConsumerRebalanceListener</code></a> for more details.</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)">subscribe</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Parameters:</span></dt><dd><code>topics</code> - The list of topics to subscribe to</dd><dd><code>listener</code> - Non-null listener instance to get notifications on partition assignment/revocation for the
+ subscribed topics</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Collection, ConsumerRebalanceListener)</code></a></dd></dl>
+</li>
+</ul>
+<a name="subscribe(java.util.Collection)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>subscribe</h4>
+<pre>public void subscribe(java.util.Collection<java.lang.String> topics)</pre>
+<div class="block">Subscribe to the given list of topics to get dynamically assigned partitions.
+ <b>Topic subscriptions are not incremental. This list will replace the current
+ assignment (if there is one).</b> It is not possible to combine topic subscription with group management
+ with manual partition assignment through <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a>.
+
+ If the given list of topics is empty, it is treated the same as <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#unsubscribe()"><code>unsubscribe()</code></a>.
+
+ <p>
+ This is a short-hand for <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Collection, ConsumerRebalanceListener)</code></a>, which
+ uses a noop listener. If you need the ability to either seek to particular offsets, you should prefer
+ <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Collection, ConsumerRebalanceListener)</code></a>, 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.</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#subscribe(java.util.Collection)">subscribe</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Parameters:</span></dt><dd><code>topics</code> - The list of topics to subscribe to</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)"><code>subscribe(Collection)</code></a></dd></dl>
+</li>
+</ul>
+<a name="subscribe(java.util.regex.Pattern, org.apache.kafka.clients.consumer.ConsumerRebalanceListener)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>subscribe</h4>
+<pre>public void subscribe(java.util.regex.Pattern pattern,
+ <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html" title="interface in org.apache.kafka.clients.consumer">ConsumerRebalanceListener</a> listener)</pre>
+<div class="block">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.
+ <p>
+ 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 -
+ <ul>
+ <li>Number of partitions change for any of the subscribed list of topics
+ <li>Topic is created or deleted
+ <li>An existing member of the consumer group dies
+ <li>A new member is added to an existing consumer group via the join API
+ </ul></div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#subscribe(java.util.regex.Pattern,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)">subscribe</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Parameters:</span></dt><dd><code>pattern</code> - Pattern to subscribe to</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.regex.Pattern,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Pattern, ConsumerRebalanceListener)</code></a></dd></dl>
+</li>
+</ul>
+<a name="unsubscribe()">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>unsubscribe</h4>
+<pre>public void unsubscribe()</pre>
+<div class="block">Unsubscribe from topics currently subscribed with <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)"><code>subscribe(Collection)</code></a>. This
+ also clears any partitions directly assigned through <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a>.</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#unsubscribe()">unsubscribe</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#unsubscribe()"><code>unsubscribe()</code></a></dd></dl>
+</li>
+</ul>
+<a name="assign(java.util.Collection)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>assign</h4>
+<pre>public void assign(java.util.Collection<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>> partitions)</pre>
+<div class="block">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).
+ <p>
+ 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 <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a>
+ and group assignment with <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)"><code>subscribe(Collection, ConsumerRebalanceListener)</code></a>.</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#assign(java.util.Collection)">assign</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Parameters:</span></dt><dd><code>partitions</code> - The list of partitions to assign this consumer</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)"><code>assign(Collection)</code></a></dd></dl>
+</li>
+</ul>
+<a name="poll(long)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>poll</h4>
+<pre>public <a href="../../../../../org/apache/kafka/clients/consumer/ConsumerRecords.html" title="class in org.apache.kafka.clients.consumer">ConsumerRecords</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>> poll(long timeout)</pre>
+<div class="block">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.
+ <p>
+ 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 <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#seek(org.apache.kafka.common.TopicPartition,%20long)"><code>seek(TopicPartition, long)</code></a> or automatically set as the last committed
+ offset for the subscribed list of partitions</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#poll(long)">poll</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Parameters:</span></dt><dd><code>timeout</code> - 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.</dd>
+<dt><span class="strong">Returns:</span></dt><dd>map of topic to records since the last fetch for the subscribed list of topics and partitions</dd>
+<dt><span class="strong">Throws:</span></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/InvalidOffsetException.html" title="class in org.apache.kafka.clients.consumer">InvalidOffsetException</a></code> - if the offset for a partition or set of
+ partitions is undefined or out of range and no offset reset policy has been configured</dd>
+<dd><code><a href="../../../../../org/apache/kafka/common/errors/WakeupException.html" title="class in org.apache.kafka.common.errors">WakeupException</a></code> - if <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#wakeup()"><code>wakeup()</code></a> is called before or while this
+ function is called</dd>
+<dd><code><a href="../../../../../org/apache/kafka/common/errors/AuthorizationException.html" title="class in org.apache.kafka.common.errors">AuthorizationException</a></code> - if caller does Read access to any of the subscribed
+ topics or to the configured groupId</dd>
+<dd><code><a href="../../../../../org/apache/kafka/common/KafkaException.html" title="class in org.apache.kafka.common">KafkaException</a></code> - 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)</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll(long)</code></a></dd></dl>
+</li>
+</ul>
+<a name="commitSync()">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>commitSync</h4>
+<pre>public void commitSync()</pre>
+<div class="block">Commit offsets returned on the last <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)"><code>poll()</code></a> for all the subscribed list of topics and partitions.
+ <p>
+ 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.
+ <p>
+ 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).</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#commitSync()">commitSync</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Throws:</span></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/CommitFailedException.html" title="class in org.apache.kafka.clients.consumer">CommitFailedException</a></code> - if the commit failed and cannot be retried.
+ This can only occur if you are using automatic group management with <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)"><code>subscribe(Collection)</code></a>,
+ or if there is an active group with the same groupId which is using group management.</dd>
+<dd><code><a href="../../../../../org/apache/kafka/common/errors/WakeupException.html" title="class in org.apache.kafka.common.errors">WakeupException</a></code> - if <a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#wakeup()"><code>wakeup()</code></a> is called before or while this
+ function is called</dd>
+<dd><code><a href="../../../../../org/apache/kafka/common/errors/AuthorizationException.html" title="class in org.apache.kafka.common.errors">AuthorizationException</a></code> - if not authorized to the topic or to the
+ configured groupId</dd>
+<dd><code><a href="../../../../../org/apache/kafka/common/KafkaException.html" title="class in org.apache.kafka.common">KafkaException</a></code> - for any other unrecoverable errors (e.g. if offset metadata
+ is too large or if the committed offset is invalid).</dd><dt><span class="strong">See Also:</span></dt><dd><a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html#commitSync()"><code>commitSync()</code></a></dd></dl>
+</li>
+</ul>
+<a name="commitSync(java.util.Map)">
+<!-- -->
+</a>
+<ul class="blockList">
+<li class="blockList">
+<h4>commitSync</h4>
+<pre>public void commitSync(java.util.Map<<a href="../../../../../org/apache/kafka/common/TopicPartition.html" title="class in org.apache.kafka.common">TopicPartition</a>,<a href="../../../../../org/apache/kafka/clients/consumer/OffsetAndMetadata.html" title="class in org.apache.kafka.clients.consumer">OffsetAndMetadata</a>> offsets)</pre>
+<div class="block">Commit the specified offsets for the specified list of topics and partitions.
+ <p>
+ 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.
+ <p>
+ 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).</div>
+<dl>
+<dt><strong>Specified by:</strong></dt>
+<dd><code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html#commitSync(java.util.Map)">commitSync</a></code> in interface <code><a href="../../../../../org/apache/kafka/clients/consumer/Consumer.html" title="interface in org.apache.kafka.clients.consumer">Consumer</a><<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">K</a>,<a href="../../../../../org/apache/kafka/clients/consumer/KafkaConsumer.html" title="type parameter in KafkaConsumer">V</a>></code></dd>
+<dt><span class="strong">Parameters:</span></dt><dd><code>offsets</code> - A map of offsets by partition with associated metadata</dd>
+<dt><span class="strong">Throws:</span></dt>
+<dd><code><a href=".
<TRUNCATED>