makeNext() {
- if (current == null || !current.hasNext()) {
+ while (current == null || !current.hasNext()) {
if (iters.hasNext())
current = iters.next().iterator();
else
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index d7c8e14cea0cb..7290a38445646 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -16,19 +16,18 @@
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;
+import org.apache.kafka.clients.consumer.internals.ConsumerInterceptors;
import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
import org.apache.kafka.clients.consumer.internals.Fetcher;
import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.internals.PartitionAssignor;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
-import org.apache.kafka.clients.consumer.internals.ConsumerInterceptors;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.metrics.JmxReporter;
import org.apache.kafka.common.metrics.MetricConfig;
@@ -36,6 +35,7 @@
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.network.ChannelBuilder;
import org.apache.kafka.common.network.Selector;
+import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.SystemTime;
@@ -45,7 +45,6 @@
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.ConcurrentModificationException;
@@ -53,6 +52,7 @@
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -76,7 +76,7 @@
* Offsets and Consumer Position
* Kafka maintains a numerical offset for each record in a partition. This offset acts as a kind of unique identifier of
* a record within that partition, and also denotes the position of the consumer in the partition. That is, a consumer
- * which has position 5 has consumed records with offsets 0 through 4 and will next receive record with offset 5. There
+ * 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.
*
* The {@link #position(TopicPartition) position} of the consumer gives the offset of the next record that will be given
@@ -95,22 +95,23 @@
*
*
Consumer Groups and Topic Subscriptions
*
- * Kafka uses the concept of consumer groups to allow a pool of processes to divide up the work of consuming and
+ * 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, as is more likely, they can be
- * distributed over many machines to provide additional scalability and fault tolerance for processing.
+ * distributed over many machines to provide scalability and fault tolerance for processing.
*
- * Each Kafka consumer is able to configure a consumer group that it belongs to, and can dynamically set the
- * list of topics it wants to subscribe to through {@link #subscribe(List, ConsumerRebalanceListener)},
- * or subscribe to all topics matching certain pattern through {@link #subscribe(Pattern, ConsumerRebalanceListener)}.
- * Kafka will deliver each message in the
- * subscribed topics to one process in each consumer group. This is achieved by balancing the partitions in the topic
- * over the consumer processes in each group. So if there is a topic with four partitions, and a consumer group with two
- * processes, each process would consume from two partitions. This group membership is maintained dynamically: if a
- * process fails the partitions assigned to it will be reassigned to other processes in the same group, and if a new
- * process joins the group, partitions will be moved from existing consumers to this new process.
+ * 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 {@link #subscribe(Collection, ConsumerRebalanceListener) 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.
*
- * So if two processes subscribe to a topic both specifying different groups they will each get all the records in that
- * topic; if they both specify the same group they will each get about half the records.
+ * 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. 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.
*
* 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
@@ -127,8 +128,51 @@
* commits (note that offsets are always committed for a given consumer group), etc.
* See Storing Offsets Outside Kafka for more details
*
- * It is also possible for the consumer to manually specify the partitions that are assigned to it through {@link #assign(List)},
- * which disables this dynamic partition assignment.
+ * It is also possible for the consumer to manually assign specific partitions
+ * (similar to the older "simple" consumer) using {@link #assign(Collection)}. In this case, dynamic partition
+ * assignment and consumer group coordination will be disabled.
+ *
+ *
+ *
+ * After subscribing to a set of topics, the consumer will automatically join the group when {@link #poll(long)} 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 session timeout 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.
+ *
+ * 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 {@link CommitFailedException} thrown from {@link #commitSync()}. 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.
+ *
+ * The consumer provides two configuration settings to control this behavior:
+ *
+ * session.timeout.ms: By increasing the session timeout, you can give the consumer more
+ * time to handle a batch of records returned from {@link #poll(long)}. 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 {@link #close()} is not impacted since
+ * the consumer will send an explicit message to the server to leave the group and cause an immediate
+ * rebalance.max.poll.records: 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.
+ *
+ * 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 {@link #pause(Collection)}
+ * 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
@@ -166,11 +210,11 @@
* In this example the client is subscribing to the topics foo and bar as part of a group of consumers
* called test as described above.
*
- * The broker will automatically detect failed processes in the test group by using a heartbeat mechanism. The
- * consumer will automatically ping the cluster periodically, which lets the cluster know that it is alive. Note that
- * the consumer is single-threaded, so periodic heartbeats can only be sent when {@link #poll(long)} 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 {@link #poll(long)} for a period of time longer than session.timeout.ms
+ * The broker will automatically detect failed processes in the test group by using a heartbeat mechanism. The
+ * consumer will automatically ping the cluster periodically, which lets the cluster know that it is alive. Note that
+ * the consumer is single-threaded, so periodic heartbeats can only be sent when {@link #poll(long)} 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 {@link #poll(long)} for a period of time longer than session.timeout.ms
* then it will be considered dead and its partitions will be assigned to another process.
*
* The deserializer settings specify how to turn bytes into objects. For example, by specifying string deserializers, we
@@ -244,27 +288,23 @@
* Note: The committed offset should always be the offset of the next message that your application will read.
* Thus, when calling {@link #commitSync(Map) commitSync(offsets)} you should add one to the offset of the last message processed.
*
- *
Subscribing To Specific Partitions
+ *
*
- * In the previous examples we subscribed to the topics we were interested in and let Kafka give our particular process
- * a fair share of the partitions for those topics. This provides a simple load balancing mechanism so multiple
- * instances of our program can divided up the work of processing records.
- *
- * In this mode the consumer will just get the partitions it subscribes to and if the consumer instance fails no attempt
- * will be made to rebalance partitions to other instances.
+ * 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:
*
- * There are several cases where this makes sense:
*
- * - The first case is if the process is maintaining some kind of local state associated with that partition (like a
- * local on-disk key-value store) and hence it should only get records for the partition it is maintaining on disk.
- *
- Another case is if the process itself is highly available and will be restarted if it fails (perhaps using a
+ *
- 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, rather the consuming process
+ * 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.
*
*
- * This mode is easy to specify, rather than subscribing to the topic, the consumer just subscribes to particular
- * partitions:
+ * To use this mode, instead of subscribing to the topic using {@link #subscribe(Collection) subscribe}, you just call
+ * {@link #assign(Collection)} with the full list of partitions that you want to consume.
*
*
* String topic = "foo";
@@ -273,11 +313,15 @@
* consumer.assign(Arrays.asList(partition0, partition1));
*
*
- * The group that the consumer specifies is still used for committing offsets, but now the set of partitions will only
- * be changed if the consumer specifies new partitions, and no attempt at failure detection will be made.
+ * Once assigned, you can call {@link #poll(long) 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 {@link #assign(Collection) 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.
*
- * It isn't possible to mix both subscription to specific partitions (with no load balancing) and to topics (with load
- * balancing) using the same consumer instance.
+ * Note that it isn't possible to mix manual partition assignment (i.e. using {@link #assign(Collection) assign})
+ * with dynamic partition assignment through topic subscription (i.e. using {@link #subscribe(Collection) subscribe}).
*
*
*
@@ -313,7 +357,7 @@
* 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
- * {@link ConsumerRebalanceListener} instance in the call to {@link #subscribe(List, ConsumerRebalanceListener)}
+ * {@link ConsumerRebalanceListener} instance in the call to {@link #subscribe(Collection, ConsumerRebalanceListener)}
* and {@link #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 {@link ConsumerRebalanceListener#onPartitionsRevoked(Collection)}. When partitions are assigned to a
@@ -342,7 +386,7 @@
*
* Kafka allows specifying the position using {@link #seek(TopicPartition, long)} to specify the new position. Special
* methods for seeking to the earliest and latest offset the server maintains are also available (
- * {@link #seekToBeginning(TopicPartition...)} and {@link #seekToEnd(TopicPartition...)} respectively).
+ * {@link #seekToBeginning(Collection)} and {@link #seekToEnd(Collection)} respectively).
*
*
Consumption Flow Control
*
@@ -359,7 +403,7 @@
* fetching other topics.
*
*
- * Kafka supports dynamic controlling of consumption flows by using {@link #pause(TopicPartition...)} and {@link #resume(TopicPartition...)}
+ * Kafka supports dynamic controlling of consumption flows by using {@link #pause(Collection)} and {@link #resume(Collection)}
* to pause the consumption on the specified assigned partitions and resume the consumption
* on the specified paused partitions respectively in the future {@link #poll(long)} calls.
*
@@ -453,7 +497,6 @@
* commit.
*
*/
-@InterfaceStability.Unstable
public class KafkaConsumer implements Consumer {
private static final Logger log = LoggerFactory.getLogger(KafkaConsumer.class);
@@ -586,8 +629,9 @@ private KafkaConsumer(ConsumerConfig config,
config.getInt(ConsumerConfig.SEND_BUFFER_CONFIG),
config.getInt(ConsumerConfig.RECEIVE_BUFFER_CONFIG),
config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), time);
- this.client = new ConsumerNetworkClient(netClient, metadata, time, retryBackoffMs);
- OffsetResetStrategy offsetResetStrategy = OffsetResetStrategy.valueOf(config.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG).toUpperCase());
+ this.client = new ConsumerNetworkClient(netClient, metadata, time, retryBackoffMs,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG));
+ OffsetResetStrategy offsetResetStrategy = OffsetResetStrategy.valueOf(config.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG).toUpperCase(Locale.ROOT));
this.subscriptions = new SubscriptionState(offsetResetStrategy);
List assignors = config.getConfiguredInstances(
ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
@@ -612,7 +656,8 @@ private KafkaConsumer(ConsumerConfig config,
new ConsumerCoordinator.DefaultOffsetCommitCallback(),
config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG),
config.getLong(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG),
- this.interceptors);
+ this.interceptors,
+ config.getBoolean(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG));
if (keyDeserializer == null) {
this.keyDeserializer = config.getConfiguredInstance(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
Deserializer.class);
@@ -659,7 +704,7 @@ private KafkaConsumer(ConsumerConfig config,
/**
* Get the set of partitions currently assigned to this consumer. If subscription happened by directly assigning
- * partitions using {@link #assign(List)} then this will simply return the same partitions that
+ * partitions using {@link #assign(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).
@@ -676,7 +721,7 @@ public Set assignment() {
/**
* Get the current subscription. Will return the same topics used in the most recent call to
- * {@link #subscribe(List, ConsumerRebalanceListener)}, or an empty set if no such call has been made.
+ * {@link #subscribe(Collection, ConsumerRebalanceListener)}, or an empty set if no such call has been made.
* @return The set of topics currently subscribed to
*/
public Set subscription() {
@@ -692,7 +737,7 @@ public Set subscription() {
* Subscribe to the given list of topics to get dynamically
* assigned partitions. Topic subscriptions are not incremental. This list will replace the current
* assignment (if there is one). Note that it is not possible to combine topic subscription with group management
- * with manual partition assignment through {@link #assign(List)}.
+ * with manual partition assignment through {@link #assign(Collection)}.
*
* If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
*
@@ -717,7 +762,7 @@ public Set subscription() {
* subscribed topics
*/
@Override
- public void subscribe(List topics, ConsumerRebalanceListener listener) {
+ public void subscribe(Collection topics, ConsumerRebalanceListener listener) {
acquire();
try {
if (topics.isEmpty()) {
@@ -737,21 +782,21 @@ public void subscribe(List topics, ConsumerRebalanceListener listener) {
* Subscribe to the given list of topics to get dynamically assigned partitions.
* Topic subscriptions are not incremental. This list will replace the current
* assignment (if there is one). It is not possible to combine topic subscription with group management
- * with manual partition assignment through {@link #assign(List)}.
+ * with manual partition assignment through {@link #assign(Collection)}.
*
* If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
*
*
- * This is a short-hand for {@link #subscribe(List, ConsumerRebalanceListener)}, which
+ * This is a short-hand for {@link #subscribe(Collection, ConsumerRebalanceListener)}, which
* uses a noop listener. If you need the ability to either seek to particular offsets, you should prefer
- * {@link #subscribe(List, ConsumerRebalanceListener)}, since group rebalances will cause partition offsets
+ * {@link #subscribe(Collection, ConsumerRebalanceListener)}, since group rebalances will cause partition offsets
* to be reset. You should also prefer to provide your own listener if you are doing your own offset
* management since the listener gives you an opportunity to commit offsets before a rebalance finishes.
*
* @param topics The list of topics to subscribe to
*/
@Override
- public void subscribe(List topics) {
+ public void subscribe(Collection topics) {
subscribe(topics, new NoOpConsumerRebalanceListener());
}
@@ -784,8 +829,8 @@ public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {
}
/**
- * Unsubscribe from topics currently subscribed with {@link #subscribe(List)}. This
- * also clears any partitions directly assigned through {@link #assign(List)}.
+ * Unsubscribe from topics currently subscribed with {@link #subscribe(Collection)}. This
+ * also clears any partitions directly assigned through {@link #assign(Collection)}.
*/
public void unsubscribe() {
acquire();
@@ -805,13 +850,13 @@ public void 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 {@link #assign(List)}
- * and group assignment with {@link #subscribe(List, ConsumerRebalanceListener)}.
+ * metadata change. Note that it is not possible to use both manual partition assignment with {@link #assign(Collection)}
+ * and group assignment with {@link #subscribe(Collection, ConsumerRebalanceListener)}.
*
* @param partitions The list of partitions to assign this consumer
*/
@Override
- public void assign(List partitions) {
+ public void assign(Collection partitions) {
acquire();
try {
log.debug("Subscribed to partition(s): {}", Utils.join(partitions, ", "));
@@ -870,7 +915,7 @@ public ConsumerRecords poll(long timeout) {
// must return these records to users to process before being interrupted or
// auto-committing offsets
fetcher.sendFetches(metadata.fetch());
- client.quickPoll();
+ client.quickPoll(false);
return this.interceptors == null
? new ConsumerRecords<>(records) : this.interceptors.onConsume(new ConsumerRecords<>(records));
}
@@ -930,7 +975,7 @@ private Map>> pollOnce(long timeout) {
* encountered (in which case it is thrown to the caller).
*
* @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
- * This can only occur if you are using automatic group management with {@link #subscribe(List)},
+ * This can only occur if you are using automatic group management with {@link #subscribe(Collection)},
* or if there is an active group with the same groupId which is using group management.
* @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
* function is called
@@ -962,7 +1007,7 @@ public void commitSync() {
*
* @param offsets A map of offsets by partition with associated metadata
* @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
- * This can only occur if you are using automatic group management with {@link #subscribe(List)},
+ * This can only occur if you are using automatic group management with {@link #subscribe(Collection)},
* or if there is an active group with the same groupId which is using group management.
* @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
* function is called
@@ -1062,11 +1107,10 @@ public void seek(TopicPartition partition, long offset) {
* first offset in all partitions only when {@link #poll(long)} or {@link #position(TopicPartition)} are called.
* If no partition is provided, seek to the first offset for all of the currently assigned partitions.
*/
- public void seekToBeginning(TopicPartition... partitions) {
+ public void seekToBeginning(Collection partitions) {
acquire();
try {
- Collection parts = partitions.length == 0 ? this.subscriptions.assignedPartitions()
- : Arrays.asList(partitions);
+ Collection parts = partitions.size() == 0 ? this.subscriptions.assignedPartitions() : partitions;
for (TopicPartition tp : parts) {
log.debug("Seeking to beginning of partition {}", tp);
subscriptions.needOffsetReset(tp, OffsetResetStrategy.EARLIEST);
@@ -1081,11 +1125,10 @@ public void seekToBeginning(TopicPartition... partitions) {
* final offset in all partitions only when {@link #poll(long)} or {@link #position(TopicPartition)} are called.
* If no partition is provided, seek to the final offset for all of the currently assigned partitions.
*/
- public void seekToEnd(TopicPartition... partitions) {
+ public void seekToEnd(Collection partitions) {
acquire();
try {
- Collection parts = partitions.length == 0 ? this.subscriptions.assignedPartitions()
- : Arrays.asList(partitions);
+ Collection parts = partitions.size() == 0 ? this.subscriptions.assignedPartitions() : partitions;
for (TopicPartition tp : parts) {
log.debug("Seeking to end of partition {}", tp);
subscriptions.needOffsetReset(tp, OffsetResetStrategy.LATEST);
@@ -1191,7 +1234,7 @@ public List partitionsFor(String topic) {
if (parts != null)
return parts;
- Map> topicMetadata = fetcher.getTopicMetadata(Collections.singletonList(topic), requestTimeoutMs);
+ Map> topicMetadata = fetcher.getTopicMetadata(new MetadataRequest(Collections.singletonList(topic)), requestTimeoutMs);
return topicMetadata.get(topic);
} finally {
release();
@@ -1221,13 +1264,13 @@ public Map> listTopics() {
/**
* Suspend fetching from the requested partitions. Future calls to {@link #poll(long)} will not return
- * any records from these partitions until they have been resumed using {@link #resume(TopicPartition...)}.
+ * any records from these partitions until they have been resumed using {@link #resume(Collection)}.
* Note that this method does not affect partition subscription. In particular, it does not cause a group
* rebalance when automatic assignment is used.
* @param partitions The partitions which should be paused
*/
@Override
- public void pause(TopicPartition... partitions) {
+ public void pause(Collection partitions) {
acquire();
try {
for (TopicPartition partition: partitions) {
@@ -1240,34 +1283,34 @@ public void pause(TopicPartition... partitions) {
}
/**
- * Get the set of partitions that were previously paused by a call to {@link #pause(TopicPartition...)}.
- *
- * @return The set of paused partitions
+ * Resume specified partitions which have been paused with {@link #pause(Collection)}. New calls to
+ * {@link #poll(long)} will return records from these partitions if there are any to be fetched.
+ * If the partitions were not previously paused, this method is a no-op.
+ * @param partitions The partitions which should be resumed
*/
@Override
- public Set paused() {
+ public void resume(Collection partitions) {
acquire();
try {
- return Collections.unmodifiableSet(subscriptions.pausedPartitions());
+ for (TopicPartition partition: partitions) {
+ log.debug("Resuming partition {}", partition);
+ subscriptions.resume(partition);
+ }
} finally {
release();
}
}
/**
- * Resume specified partitions which have been paused with {@link #pause(TopicPartition...)}. New calls to
- * {@link #poll(long)} will return records from these partitions if there are any to be fetched.
- * If the partitions were not previously paused, this method is a no-op.
- * @param partitions The partitions which should be resumed
+ * Get the set of partitions that were previously paused by a call to {@link #pause(Collection)}.
+ *
+ * @return The set of paused partitions
*/
@Override
- public void resume(TopicPartition... partitions) {
+ public Set paused() {
acquire();
try {
- for (TopicPartition partition: partitions) {
- log.debug("Resuming partition {}", partition);
- subscriptions.resume(partition);
- }
+ return Collections.unmodifiableSet(subscriptions.pausedPartitions());
} finally {
release();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
index c7f0a46754b35..8dce1f1f9411c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
@@ -67,7 +67,7 @@ public MockConsumer(OffsetResetStrategy offsetResetStrategy) {
this.exception = null;
this.wakeup = new AtomicBoolean(false);
}
-
+
@Override
public Set assignment() {
return this.subscriptions.assignedPartitions();
@@ -86,7 +86,7 @@ public Set subscription() {
}
@Override
- public void subscribe(List topics) {
+ public void subscribe(Collection topics) {
subscribe(topics, new NoOpConsumerRebalanceListener());
}
@@ -105,13 +105,13 @@ public void subscribe(Pattern pattern, final ConsumerRebalanceListener listener)
}
@Override
- public void subscribe(List topics, final ConsumerRebalanceListener listener) {
+ public void subscribe(Collection topics, final ConsumerRebalanceListener listener) {
ensureNotClosed();
this.subscriptions.subscribe(topics, listener);
}
@Override
- public void assign(List partitions) {
+ public void assign(Collection partitions) {
ensureNotClosed();
this.subscriptions.assignFromUser(partitions);
}
@@ -238,7 +238,7 @@ public long position(TopicPartition partition) {
}
@Override
- public void seekToBeginning(TopicPartition... partitions) {
+ public void seekToBeginning(Collection partitions) {
ensureNotClosed();
for (TopicPartition tp : partitions)
subscriptions.needOffsetReset(tp, OffsetResetStrategy.EARLIEST);
@@ -249,7 +249,7 @@ public void updateBeginningOffsets(Map newOffsets) {
}
@Override
- public void seekToEnd(TopicPartition... partitions) {
+ public void seekToEnd(Collection partitions) {
ensureNotClosed();
for (TopicPartition tp : partitions)
subscriptions.needOffsetReset(tp, OffsetResetStrategy.LATEST);
@@ -287,7 +287,7 @@ public void updatePartitions(String topic, List partitions) {
}
@Override
- public void pause(TopicPartition... partitions) {
+ public void pause(Collection partitions) {
for (TopicPartition partition : partitions) {
subscriptions.pause(partition);
paused.add(partition);
@@ -295,7 +295,7 @@ public void pause(TopicPartition... partitions) {
}
@Override
- public void resume(TopicPartition... partitions) {
+ public void resume(Collection partitions) {
for (TopicPartition partition : partitions) {
subscriptions.resume(partition);
paused.remove(partition);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java
index 66b257dba3099..df8bf37adbada 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndMetadata.java
@@ -29,9 +29,6 @@ public class OffsetAndMetadata implements Serializable {
* @param metadata Non-null metadata
*/
public OffsetAndMetadata(long offset, String metadata) {
- if (metadata == null)
- throw new IllegalArgumentException("Metadata cannot be null");
-
this.offset = offset;
this.metadata = metadata;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index c6492bc66d2a6..15185d7de03a1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -17,6 +17,7 @@
import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
+import org.apache.kafka.common.errors.GroupCoordinatorNotAvailableException;
import org.apache.kafka.common.errors.IllegalGenerationException;
import org.apache.kafka.common.errors.RebalanceInProgressException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
@@ -174,7 +175,7 @@ protected abstract void onJoinComplete(int generation,
*/
public void ensureCoordinatorKnown() {
while (coordinatorUnknown()) {
- RequestFuture future = sendGroupMetadataRequest();
+ RequestFuture future = sendGroupCoordinatorRequest();
client.poll(future);
if (future.failed()) {
@@ -216,14 +217,26 @@ public void ensureActiveGroup() {
continue;
}
- RequestFuture future = performGroupJoin();
+ RequestFuture future = sendJoinGroupRequest();
+ future.addListener(new RequestFutureListener() {
+ @Override
+ public void onSuccess(ByteBuffer value) {
+ // handle join completion in the callback so that the callback will be invoked
+ // even if the consumer is woken up before finishing the rebalance
+ onJoinComplete(generation, memberId, protocol, value);
+ needsJoinPrepare = true;
+ heartbeatTask.reset();
+ }
+
+ @Override
+ public void onFailure(RuntimeException e) {
+ // we handle failures below after the request finishes. if the join completes
+ // after having been woken up, the exception is ignored and we will rejoin
+ }
+ });
client.poll(future);
- if (future.succeeded()) {
- onJoinComplete(generation, memberId, protocol, future.value());
- needsJoinPrepare = true;
- heartbeatTask.reset();
- } else {
+ if (future.failed()) {
RuntimeException exception = future.exception();
if (exception instanceof UnknownMemberIdException ||
exception instanceof RebalanceInProgressException ||
@@ -299,12 +312,12 @@ public void onFailure(RuntimeException e) {
* elected leader by the coordinator.
* @return A request future which wraps the assignment returned from the group leader
*/
- private RequestFuture performGroupJoin() {
+ private RequestFuture sendJoinGroupRequest() {
if (coordinatorUnknown())
return RequestFuture.coordinatorNotAvailable();
// send a join group request to the coordinator
- log.debug("(Re-)joining group {}", groupId);
+ log.info("(Re-)joining group {}", groupId);
JoinGroupRequest request = new JoinGroupRequest(
groupId,
this.sessionTimeoutMs,
@@ -312,8 +325,7 @@ private RequestFuture performGroupJoin() {
protocolType(),
metadata());
- // create the request for the coordinator
- log.debug("Issuing request ({}: {}) to coordinator {}", ApiKeys.JOIN_GROUP, request, this.coordinator.id());
+ log.debug("Sending JoinGroup ({}) to coordinator {}", request, this.coordinator);
return client.send(coordinator, ApiKeys.JOIN_GROUP, request)
.compose(new JoinGroupResponseHandler());
}
@@ -328,10 +340,9 @@ public JoinGroupResponse parse(ClientResponse response) {
@Override
public void handle(JoinGroupResponse joinResponse, RequestFuture future) {
- // process the response
- short errorCode = joinResponse.errorCode();
- if (errorCode == Errors.NONE.code()) {
- log.debug("Joined group: {}", joinResponse.toStruct());
+ Errors error = Errors.forCode(joinResponse.errorCode());
+ if (error == Errors.NONE) {
+ log.debug("Received successful join group response for group {}: {}", groupId, joinResponse.toStruct());
AbstractCoordinator.this.memberId = joinResponse.memberId();
AbstractCoordinator.this.generation = joinResponse.generationId();
AbstractCoordinator.this.rejoinNeeded = false;
@@ -342,37 +353,33 @@ public void handle(JoinGroupResponse joinResponse, RequestFuture fut
} else {
onJoinFollower().chain(future);
}
- } else if (errorCode == Errors.GROUP_LOAD_IN_PROGRESS.code()) {
- log.debug("Attempt to join group {} rejected since coordinator is loading the group.", groupId);
+ } else if (error == Errors.GROUP_LOAD_IN_PROGRESS) {
+ log.debug("Attempt to join group {} rejected since coordinator {} is loading the group.", groupId,
+ coordinator);
// backoff and retry
- future.raise(Errors.forCode(errorCode));
- } else if (errorCode == Errors.UNKNOWN_MEMBER_ID.code()) {
+ future.raise(error);
+ } else if (error == Errors.UNKNOWN_MEMBER_ID) {
// reset the member id and retry immediately
AbstractCoordinator.this.memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID;
- log.info("Attempt to join group {} failed due to unknown member id, resetting and retrying.",
- groupId);
+ log.debug("Attempt to join group {} failed due to unknown member id.", groupId);
future.raise(Errors.UNKNOWN_MEMBER_ID);
- } else if (errorCode == Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()
- || errorCode == Errors.NOT_COORDINATOR_FOR_GROUP.code()) {
+ } else if (error == Errors.GROUP_COORDINATOR_NOT_AVAILABLE
+ || error == Errors.NOT_COORDINATOR_FOR_GROUP) {
// re-discover the coordinator and retry with backoff
coordinatorDead();
- log.info("Attempt to join group {} failed due to obsolete coordinator information, retrying.",
- groupId);
- future.raise(Errors.forCode(errorCode));
- } else if (errorCode == Errors.INCONSISTENT_GROUP_PROTOCOL.code()
- || errorCode == Errors.INVALID_SESSION_TIMEOUT.code()
- || errorCode == Errors.INVALID_GROUP_ID.code()) {
+ log.debug("Attempt to join group {} failed due to obsolete coordinator information: {}", groupId, error.message());
+ future.raise(error);
+ } else if (error == Errors.INCONSISTENT_GROUP_PROTOCOL
+ || error == Errors.INVALID_SESSION_TIMEOUT
+ || error == Errors.INVALID_GROUP_ID) {
// log the error and re-throw the exception
- Errors error = Errors.forCode(errorCode);
- log.error("Attempt to join group {} failed due to: {}",
- groupId, error.exception().getMessage());
+ log.error("Attempt to join group {} failed due to fatal error: {}", groupId, error.message());
future.raise(error);
- } else if (errorCode == Errors.GROUP_AUTHORIZATION_FAILED.code()) {
+ } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
future.raise(new GroupAuthorizationException(groupId));
} else {
// unexpected error, throw the exception
- future.raise(new KafkaException("Unexpected error in join group response: "
- + Errors.forCode(joinResponse.errorCode()).exception().getMessage()));
+ future.raise(new KafkaException("Unexpected error in join group response: " + error.message()));
}
}
}
@@ -381,7 +388,7 @@ private RequestFuture onJoinFollower() {
// send follower's sync group with an empty assignment
SyncGroupRequest request = new SyncGroupRequest(groupId, generation,
memberId, Collections.emptyMap());
- log.debug("Issuing follower SyncGroup ({}: {}) to coordinator {}", ApiKeys.SYNC_GROUP, request, this.coordinator.id());
+ log.debug("Sending follower SyncGroup for group {} to coordinator {}: {}", groupId, this.coordinator, request);
return sendSyncGroupRequest(request);
}
@@ -392,7 +399,7 @@ private RequestFuture onJoinLeader(JoinGroupResponse joinResponse) {
joinResponse.members());
SyncGroupRequest request = new SyncGroupRequest(groupId, generation, memberId, groupAssignment);
- log.debug("Issuing leader SyncGroup ({}: {}) to coordinator {}", ApiKeys.SYNC_GROUP, request, this.coordinator.id());
+ log.debug("Sending leader SyncGroup for group {} to coordinator {}: {}", groupId, this.coordinator, request);
return sendSyncGroupRequest(request);
} catch (RuntimeException e) {
return RequestFuture.failure(e);
@@ -403,10 +410,10 @@ private RequestFuture sendSyncGroupRequest(SyncGroupRequest request)
if (coordinatorUnknown())
return RequestFuture.coordinatorNotAvailable();
return client.send(coordinator, ApiKeys.SYNC_GROUP, request)
- .compose(new SyncGroupRequestHandler());
+ .compose(new SyncGroupResponseHandler());
}
- private class SyncGroupRequestHandler extends CoordinatorResponseHandler {
+ private class SyncGroupResponseHandler extends CoordinatorResponseHandler {
@Override
public SyncGroupResponse parse(ClientResponse response) {
@@ -418,7 +425,7 @@ public void handle(SyncGroupResponse syncResponse,
RequestFuture future) {
Errors error = Errors.forCode(syncResponse.errorCode());
if (error == Errors.NONE) {
- log.debug("Received successful sync group response for group {}: {}", groupId, syncResponse.toStruct());
+ log.info("Successfully joined group {} with generation {}", groupId, generation);
sensors.syncLatency.record(response.requestLatencyMs());
future.complete(syncResponse.memberAssignment());
} else {
@@ -426,20 +433,20 @@ public void handle(SyncGroupResponse syncResponse,
if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
future.raise(new GroupAuthorizationException(groupId));
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
- log.info("SyncGroup for group {} failed due to coordinator rebalance, rejoining the group", groupId);
+ log.debug("SyncGroup for group {} failed due to coordinator rebalance", groupId);
future.raise(error);
} else if (error == Errors.UNKNOWN_MEMBER_ID
|| error == Errors.ILLEGAL_GENERATION) {
- log.info("SyncGroup for group {} failed due to {}, rejoining the group", groupId, error);
+ log.debug("SyncGroup for group {} failed due to {}", groupId, error);
AbstractCoordinator.this.memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID;
future.raise(error);
} else if (error == Errors.GROUP_COORDINATOR_NOT_AVAILABLE
|| error == Errors.NOT_COORDINATOR_FOR_GROUP) {
- log.info("SyncGroup for group {} failed due to {}, will find new coordinator and rejoin", groupId, error);
+ log.debug("SyncGroup for group {} failed due to {}", groupId, error);
coordinatorDead();
future.raise(error);
} else {
- future.raise(new KafkaException("Unexpected error from SyncGroup: " + error.exception().getMessage()));
+ future.raise(new KafkaException("Unexpected error from SyncGroup: " + error.message()));
}
}
}
@@ -450,7 +457,7 @@ public void handle(SyncGroupResponse syncResponse,
* one of the brokers. The returned future should be polled to get the result of the request.
* @return A request future which indicates the completion of the metadata request
*/
- private RequestFuture sendGroupMetadataRequest() {
+ private RequestFuture sendGroupCoordinatorRequest() {
// initiate the group metadata request
// find a node to ask about the coordinator
Node node = this.client.leastLoadedNode();
@@ -460,7 +467,7 @@ private RequestFuture sendGroupMetadataRequest() {
return RequestFuture.noBrokersAvailable();
} else {
// create a group metadata request
- log.debug("Issuing group metadata request to broker {}", node.id());
+ log.debug("Sending coordinator request for group {} to broker {}", groupId, node);
GroupCoordinatorRequest metadataRequest = new GroupCoordinatorRequest(this.groupId);
return client.send(node, ApiKeys.GROUP_COORDINATOR, metadataRequest)
.compose(new RequestFutureAdapter() {
@@ -473,7 +480,7 @@ public void onSuccess(ClientResponse response, RequestFuture future) {
}
private void handleGroupMetadataResponse(ClientResponse resp, RequestFuture future) {
- log.debug("Group metadata response {}", resp);
+ log.debug("Received group coordinator response {}", resp);
if (!coordinatorUnknown()) {
// We already found the coordinator, so ignore the request
@@ -483,22 +490,24 @@ private void handleGroupMetadataResponse(ClientResponse resp, RequestFuture 0)
heartbeatTask.reset();
future.complete(null);
- } else if (errorCode == Errors.GROUP_AUTHORIZATION_FAILED.code()) {
+ } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
future.raise(new GroupAuthorizationException(groupId));
} else {
- future.raise(Errors.forCode(errorCode));
+ future.raise(error);
}
}
}
@@ -524,7 +533,8 @@ public boolean coordinatorUnknown() {
*/
protected void coordinatorDead() {
if (this.coordinator != null) {
- log.info("Marking the coordinator {} dead.", this.coordinator.id());
+ log.info("Marking the coordinator {} dead for group {}", this.coordinator, groupId);
+ client.failUnsentRequests(this.coordinator, GroupCoordinatorNotAvailableException.INSTANCE);
this.coordinator = null;
}
}
@@ -566,7 +576,7 @@ public void onSuccess(Void value) {}
@Override
public void onFailure(RuntimeException e) {
- log.info("LeaveGroup request failed with error", e);
+ log.debug("LeaveGroup request for group {} failed with error", groupId, e);
}
});
@@ -608,33 +618,33 @@ public HeartbeatResponse parse(ClientResponse response) {
@Override
public void handle(HeartbeatResponse heartbeatResponse, RequestFuture future) {
sensors.heartbeatLatency.record(response.requestLatencyMs());
- short errorCode = heartbeatResponse.errorCode();
- if (errorCode == Errors.NONE.code()) {
- log.debug("Received successful heartbeat response.");
+ Errors error = Errors.forCode(heartbeatResponse.errorCode());
+ if (error == Errors.NONE) {
+ log.debug("Received successful heartbeat response for group {}", groupId);
future.complete(null);
- } else if (errorCode == Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code()
- || errorCode == Errors.NOT_COORDINATOR_FOR_GROUP.code()) {
- log.info("Attempt to heart beat failed since coordinator is either not started or not valid, marking it as dead.");
+ } else if (error == Errors.GROUP_COORDINATOR_NOT_AVAILABLE
+ || error == Errors.NOT_COORDINATOR_FOR_GROUP) {
+ log.debug("Attempt to heart beat failed for group {} since coordinator {} is either not started or not valid.",
+ groupId, coordinator);
coordinatorDead();
- future.raise(Errors.forCode(errorCode));
- } else if (errorCode == Errors.REBALANCE_IN_PROGRESS.code()) {
- log.info("Attempt to heart beat failed since the group is rebalancing, try to re-join group.");
+ future.raise(error);
+ } else if (error == Errors.REBALANCE_IN_PROGRESS) {
+ log.debug("Attempt to heart beat failed for group {} since it is rebalancing.", groupId);
AbstractCoordinator.this.rejoinNeeded = true;
future.raise(Errors.REBALANCE_IN_PROGRESS);
- } else if (errorCode == Errors.ILLEGAL_GENERATION.code()) {
- log.info("Attempt to heart beat failed since generation id is not legal, try to re-join group.");
+ } else if (error == Errors.ILLEGAL_GENERATION) {
+ log.debug("Attempt to heart beat failed for group {} since generation id is not legal.", groupId);
AbstractCoordinator.this.rejoinNeeded = true;
future.raise(Errors.ILLEGAL_GENERATION);
- } else if (errorCode == Errors.UNKNOWN_MEMBER_ID.code()) {
- log.info("Attempt to heart beat failed since member id is not valid, reset it and try to re-join group.");
+ } else if (error == Errors.UNKNOWN_MEMBER_ID) {
+ log.debug("Attempt to heart beat failed for group {} since member id is not valid.", groupId);
memberId = JoinGroupRequest.UNKNOWN_MEMBER_ID;
AbstractCoordinator.this.rejoinNeeded = true;
future.raise(Errors.UNKNOWN_MEMBER_ID);
- } else if (errorCode == Errors.GROUP_AUTHORIZATION_FAILED.code()) {
+ } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
future.raise(new GroupAuthorizationException(groupId));
} else {
- future.raise(new KafkaException("Unexpected errorCode in heartbeat response: "
- + Errors.forCode(errorCode).exception().getMessage()));
+ future.raise(new KafkaException("Unexpected error in heartbeat response: " + error.message()));
}
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index aa39e11929116..887f47c1cde4c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -26,6 +26,7 @@
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.internals.TopicConstants;
import org.apache.kafka.common.metrics.Measurable;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
@@ -62,13 +63,16 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
private final List assignors;
private final org.apache.kafka.clients.Metadata metadata;
- private final MetadataSnapshot metadataSnapshot;
private final ConsumerCoordinatorMetrics sensors;
private final SubscriptionState subscriptions;
private final OffsetCommitCallback defaultOffsetCommitCallback;
private final boolean autoCommitEnabled;
private final AutoCommitTask autoCommitTask;
- private final ConsumerInterceptors interceptors;
+ private final ConsumerInterceptors interceptors;
+ private final boolean excludeInternalTopics;
+
+ private MetadataSnapshot metadataSnapshot;
+ private MetadataSnapshot assignmentSnapshot;
/**
* Initialize the coordination manager.
@@ -87,7 +91,8 @@ public ConsumerCoordinator(ConsumerNetworkClient client,
OffsetCommitCallback defaultOffsetCommitCallback,
boolean autoCommitEnabled,
long autoCommitIntervalMs,
- ConsumerInterceptors interceptors) {
+ ConsumerInterceptors interceptors,
+ boolean excludeInternalTopics) {
super(client,
groupId,
sessionTimeoutMs,
@@ -99,7 +104,7 @@ public ConsumerCoordinator(ConsumerNetworkClient client,
this.metadata = metadata;
this.metadata.requestUpdate();
- this.metadataSnapshot = new MetadataSnapshot();
+ this.metadataSnapshot = new MetadataSnapshot(subscriptions, metadata.fetch());
this.subscriptions = subscriptions;
this.defaultOffsetCommitCallback = defaultOffsetCommitCallback;
this.autoCommitEnabled = autoCommitEnabled;
@@ -107,9 +112,16 @@ public ConsumerCoordinator(ConsumerNetworkClient client,
addMetadataListener();
- this.autoCommitTask = autoCommitEnabled ? new AutoCommitTask(autoCommitIntervalMs) : null;
+ if (autoCommitEnabled) {
+ this.autoCommitTask = new AutoCommitTask(autoCommitIntervalMs);
+ this.autoCommitTask.reschedule();
+ } else {
+ this.autoCommitTask = null;
+ }
+
this.sensors = new ConsumerCoordinatorMetrics(metrics, metricGrpPrefix);
this.interceptors = interceptors;
+ this.excludeInternalTopics = excludeInternalTopics;
}
@Override
@@ -140,7 +152,8 @@ public void onMetadataUpdate(Cluster cluster) {
final List topicsToSubscribe = new ArrayList<>();
for (String topic : cluster.topics())
- if (subscriptions.getSubscribedPattern().matcher(topic).matches())
+ if (subscriptions.getSubscribedPattern().matcher(topic).matches() &&
+ !(excludeInternalTopics && TopicConstants.INTERNAL_TOPICS.contains(topic)))
topicsToSubscribe.add(topic);
subscriptions.changeSubscription(topicsToSubscribe);
@@ -148,8 +161,14 @@ public void onMetadataUpdate(Cluster cluster) {
}
// check if there are any changes to the metadata which should trigger a rebalance
- if (metadataSnapshot.update(subscriptions, cluster) && subscriptions.partitionsAutoAssigned())
- subscriptions.needReassignment();
+ if (subscriptions.partitionsAutoAssigned()) {
+ MetadataSnapshot snapshot = new MetadataSnapshot(subscriptions, cluster);
+ if (!snapshot.equals(metadataSnapshot)) {
+ metadataSnapshot = snapshot;
+ subscriptions.needReassignment();
+ }
+ }
+
}
});
}
@@ -167,6 +186,13 @@ protected void onJoinComplete(int generation,
String memberId,
String assignmentStrategy,
ByteBuffer assignmentBuffer) {
+ // if we were the assignor, then we need to make sure that there have been no metadata updates
+ // since the rebalance begin. Otherwise, we won't rebalance again until the next metadata change
+ if (assignmentSnapshot != null && !assignmentSnapshot.equals(metadataSnapshot)) {
+ subscriptions.needReassignment();
+ return;
+ }
+
PartitionAssignor assignor = lookupAssignor(assignmentStrategy);
if (assignor == null)
throw new IllegalStateException("Coordinator selected invalid assignment protocol: " + assignmentStrategy);
@@ -182,21 +208,21 @@ protected void onJoinComplete(int generation,
// give the assignor a chance to update internal state based on the received assignment
assignor.onAssignment(assignment);
- // restart the autocommit task if needed
+ // reschedule the auto commit starting from now
if (autoCommitEnabled)
- autoCommitTask.enable();
+ autoCommitTask.reschedule();
// execute the user's callback after rebalance
ConsumerRebalanceListener listener = subscriptions.listener();
- log.debug("Setting newly assigned partitions {}", subscriptions.assignedPartitions());
+ log.info("Setting newly assigned partitions {} for group {}", subscriptions.assignedPartitions(), groupId);
try {
Set assigned = new HashSet<>(subscriptions.assignedPartitions());
listener.onPartitionsAssigned(assigned);
} catch (WakeupException e) {
throw e;
} catch (Exception e) {
- log.error("User provided listener " + listener.getClass().getName()
- + " failed on partition assignment: ", e);
+ log.error("User provided listener {} for group {} failed on partition assignment",
+ listener.getClass().getName(), groupId, e);
}
}
@@ -220,13 +246,18 @@ protected Map performAssignment(String leaderId,
// which ensures that all metadata changes will eventually be seen
this.subscriptions.groupSubscribe(allSubscribedTopics);
metadata.setTopics(this.subscriptions.groupSubscription());
+
+ // update metadata (if needed) and keep track of the metadata used for assignment so that
+ // we can check after rebalance completion whether anything has changed
client.ensureFreshMetadata();
+ assignmentSnapshot = metadataSnapshot;
- log.debug("Performing {} assignment for subscriptions {}", assignor.name(), subscriptions);
+ log.debug("Performing assignment for group {} using strategy {} with subscriptions {}",
+ groupId, assignor.name(), subscriptions);
Map assignment = assignor.assign(metadata.fetch(), subscriptions);
- log.debug("Finished assignment: {}", assignment);
+ log.debug("Finished assignment for group {}: {}", groupId, assignment);
Map groupAssignment = new HashMap<>();
for (Map.Entry assignmentEntry : assignment.entrySet()) {
@@ -244,17 +275,18 @@ protected void onJoinPrepare(int generation, String memberId) {
// execute the user's callback before rebalance
ConsumerRebalanceListener listener = subscriptions.listener();
- log.debug("Revoking previously assigned partitions {}", subscriptions.assignedPartitions());
+ log.info("Revoking previously assigned partitions {} for group {}", subscriptions.assignedPartitions(), groupId);
try {
Set revoked = new HashSet<>(subscriptions.assignedPartitions());
listener.onPartitionsRevoked(revoked);
} catch (WakeupException e) {
throw e;
} catch (Exception e) {
- log.error("User provided listener " + listener.getClass().getName()
- + " failed on partition revocation: ", e);
+ log.error("User provided listener {} for group {} failed on partition revocation",
+ listener.getClass().getName(), groupId, e);
}
+ assignmentSnapshot = null;
subscriptions.needReassignment();
}
@@ -339,6 +371,10 @@ public void onFailure(RuntimeException e) {
cb.onComplete(offsets, e);
}
});
+
+ // ensure commit has a chance to be transmitted (without blocking on its completion)
+ // note that we allow delayed tasks to be executed in case heartbeats need to be sent
+ client.quickPoll(true);
}
/**
@@ -374,59 +410,40 @@ public void commitOffsetsSync(Map offsets) {
private class AutoCommitTask implements DelayedTask {
private final long interval;
- private boolean enabled = false;
- private boolean requestInFlight = false;
public AutoCommitTask(long interval) {
this.interval = interval;
}
- public void enable() {
- if (!enabled) {
- // there shouldn't be any instances scheduled, but call unschedule anyway to ensure
- // that this task is only ever scheduled once
- client.unschedule(this);
- this.enabled = true;
-
- if (!requestInFlight) {
- long now = time.milliseconds();
- client.schedule(this, interval + now);
- }
- }
- }
-
- public void disable() {
- this.enabled = false;
- client.unschedule(this);
+ private void reschedule() {
+ client.schedule(this, time.milliseconds() + interval);
}
private void reschedule(long at) {
- if (enabled)
- client.schedule(this, at);
+ client.schedule(this, at);
}
public void run(final long now) {
- if (!enabled)
+ if (coordinatorUnknown()) {
+ log.debug("Cannot auto-commit offsets for group {} since the coordinator is unknown", groupId);
+ reschedule(now + retryBackoffMs);
return;
+ }
- if (coordinatorUnknown()) {
- log.debug("Cannot auto-commit offsets now since the coordinator is unknown, will retry after backoff");
- client.schedule(this, now + retryBackoffMs);
+ if (needRejoin()) {
+ // skip the commit when we're rejoining since we'll commit offsets synchronously
+ // before the revocation callback is invoked
+ reschedule(now + interval);
return;
}
- requestInFlight = true;
commitOffsetsAsync(subscriptions.allConsumed(), new OffsetCommitCallback() {
@Override
public void onComplete(Map offsets, Exception exception) {
- requestInFlight = false;
if (exception == null) {
reschedule(now + interval);
- } else if (exception instanceof SendFailedException) {
- log.debug("Failed to send automatic offset commit, will retry immediately");
- reschedule(now);
} else {
- log.warn("Auto offset commit failed: {}", exception.getMessage());
+ log.warn("Auto offset commit failed for group {}: {}", groupId, exception.getMessage());
reschedule(now + interval);
}
}
@@ -436,10 +453,6 @@ public void onComplete(Map offsets, Exception
private void maybeAutoCommitOffsetsSync() {
if (autoCommitEnabled) {
- // disable periodic commits prior to committing synchronously. note that they will
- // be re-enabled after a rebalance completes
- autoCommitTask.disable();
-
try {
commitOffsetsSync(subscriptions.allConsumed());
} catch (WakeupException e) {
@@ -447,7 +460,7 @@ private void maybeAutoCommitOffsetsSync() {
throw e;
} catch (Exception e) {
// consistent with async auto-commit failures, we do not propagate the exception
- log.warn("Auto offset commit failed: ", e.getMessage());
+ log.warn("Auto offset commit failed for group {}: {}", groupId, e.getMessage());
}
}
}
@@ -481,7 +494,7 @@ private RequestFuture sendOffsetCommitRequest(final Map futu
Errors error = Errors.forCode(entry.getValue());
if (error == Errors.NONE) {
- log.debug("Committed offset {} for partition {}", offset, tp);
+ log.debug("Group {} committed offset {} for partition {}", groupId, offset, tp);
if (subscriptions.isAssigned(tp))
// update the local cache only if the partition is still assigned
subscriptions.committed(tp, offsetAndMetadata);
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
- log.error("Unauthorized to commit for group {}", groupId);
+ log.error("Not authorized to commit offsets for group {}", groupId);
future.raise(new GroupAuthorizationException(groupId));
return;
} else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
@@ -533,18 +546,18 @@ public void handle(OffsetCommitResponse commitResponse, RequestFuture futu
} else if (error == Errors.OFFSET_METADATA_TOO_LARGE
|| error == Errors.INVALID_COMMIT_OFFSET_SIZE) {
// raise the error to the user
- log.info("Offset commit for group {} failed on partition {} due to {}, will retry", groupId, tp, error);
+ log.debug("Offset commit for group {} failed on partition {}: {}", groupId, tp, error.message());
future.raise(error);
return;
} else if (error == Errors.GROUP_LOAD_IN_PROGRESS) {
// just retry
- log.info("Offset commit for group {} failed due to {}, will retry", groupId, error);
+ log.debug("Offset commit for group {} failed: {}", groupId, error.message());
future.raise(error);
return;
} else if (error == Errors.GROUP_COORDINATOR_NOT_AVAILABLE
|| error == Errors.NOT_COORDINATOR_FOR_GROUP
|| error == Errors.REQUEST_TIMED_OUT) {
- log.info("Offset commit for group {} failed due to {}, will find new coordinator and retry", groupId, error);
+ log.debug("Offset commit for group {} failed: {}", groupId, error.message());
coordinatorDead();
future.raise(error);
return;
@@ -552,19 +565,24 @@ public void handle(OffsetCommitResponse commitResponse, RequestFuture futu
|| error == Errors.ILLEGAL_GENERATION
|| error == Errors.REBALANCE_IN_PROGRESS) {
// need to re-join group
- log.error("Error {} occurred while committing offsets for group {}", error, groupId);
+ log.debug("Offset commit for group {} failed: {}", groupId, error.message());
subscriptions.needReassignment();
- future.raise(new CommitFailedException("Commit cannot be completed due to group rebalance"));
+ future.raise(new CommitFailedException("Commit cannot be completed since the group has already " +
+ "rebalanced and assigned the partitions to another member. This means that the time " +
+ "between subsequent calls to poll() was longer than the configured session.timeout.ms, " +
+ "which typically implies that the poll loop is spending too much time message processing. " +
+ "You can address this either by increasing the session timeout or by reducing the maximum " +
+ "size of batches returned in poll() with max.poll.records."));
return;
} else {
- log.error("Error committing partition {} at offset {}: {}", tp, offset, error.exception().getMessage());
- future.raise(new KafkaException("Unexpected error in commit: " + error.exception().getMessage()));
+ log.error("Group {} failed to commit partition {} at offset {}: {}", groupId, tp, offset, error.message());
+ future.raise(new KafkaException("Unexpected error in commit: " + error.message()));
return;
}
}
if (!unauthorizedTopics.isEmpty()) {
- log.error("Unauthorized to commit to topics {}", unauthorizedTopics);
+ log.error("Not authorized to commit to topics {} for group {}", unauthorizedTopics, groupId);
future.raise(new TopicAuthorizationException(unauthorizedTopics));
} else {
future.complete(null);
@@ -583,9 +601,9 @@ private RequestFuture> sendOffsetFetchReq
if (coordinatorUnknown())
return RequestFuture.coordinatorNotAvailable();
- log.debug("Fetching committed offsets for partitions: {}", partitions);
+ log.debug("Group {} fetching committed offsets for partitions: {}", groupId, partitions);
// construct the request
- OffsetFetchRequest request = new OffsetFetchRequest(this.groupId, new ArrayList(partitions));
+ OffsetFetchRequest request = new OffsetFetchRequest(this.groupId, new ArrayList<>(partitions));
// send the request with a callback
return client.send(coordinator, ApiKeys.OFFSET_FETCH, request)
@@ -606,31 +624,30 @@ public void handle(OffsetFetchResponse response, RequestFuture= 0) {
// record the position with the offset (-1 indicates no committed offset to fetch)
offsets.put(tp, new OffsetAndMetadata(data.offset, data.metadata));
} else {
- log.debug("No committed offset for partition " + tp);
+ log.debug("Group {} has no committed offset for partition {}", groupId, tp);
}
}
@@ -672,19 +689,26 @@ public double measure(MetricConfig config, long now) {
}
private static class MetadataSnapshot {
- private Map partitionsPerTopic = new HashMap<>();
+ private final Map partitionsPerTopic;
- public boolean update(SubscriptionState subscription, Cluster cluster) {
+ public MetadataSnapshot(SubscriptionState subscription, Cluster cluster) {
Map partitionsPerTopic = new HashMap<>();
for (String topic : subscription.groupSubscription())
partitionsPerTopic.put(topic, cluster.partitionCountForTopic(topic));
+ this.partitionsPerTopic = partitionsPerTopic;
+ }
- if (!partitionsPerTopic.equals(this.partitionsPerTopic)) {
- this.partitionsPerTopic = partitionsPerTopic;
- return true;
- }
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ MetadataSnapshot that = (MetadataSnapshot) o;
+ return partitionsPerTopic != null ? partitionsPerTopic.equals(that.partitionsPerTopic) : that.partitionsPerTopic == null;
+ }
- return false;
+ @Override
+ public int hashCode() {
+ return partitionsPerTopic != null ? partitionsPerTopic.hashCode() : 0;
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
index 4492306fd3a8e..d4c26568fdb01 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
@@ -19,6 +19,7 @@
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.DisconnectException;
+import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
@@ -40,12 +41,6 @@
/**
* Higher level consumer access to the network layer with basic support for futures and
* task scheduling. NOT thread-safe!
- *
- * TODO: The current implementation is simplistic in that it provides a facility for queueing requests
- * prior to delivery, but it makes no effort to retry requests which cannot be sent at the time
- * {@link #poll(long)} is called. This makes the behavior of the queue predictable and easy to
- * understand, but there are opportunities to provide timeout or retry capabilities in the future.
- * How we do this may depend on KAFKA-2120, so for now, we retain the simplistic behavior.
*/
public class ConsumerNetworkClient implements Closeable {
private static final Logger log = LoggerFactory.getLogger(ConsumerNetworkClient.class);
@@ -57,17 +52,20 @@ public class ConsumerNetworkClient implements Closeable {
private final Metadata metadata;
private final Time time;
private final long retryBackoffMs;
+ private final long unsentExpiryMs;
// wakeup enabled flag need to be volatile since it is allowed to be accessed concurrently
volatile private boolean wakeupsEnabled = true;
public ConsumerNetworkClient(KafkaClient client,
Metadata metadata,
Time time,
- long retryBackoffMs) {
+ long retryBackoffMs,
+ long requestTimeoutMs) {
this.client = client;
this.metadata = metadata;
this.time = time;
this.retryBackoffMs = retryBackoffMs;
+ this.unsentExpiryMs = requestTimeoutMs;
}
/**
@@ -140,7 +138,7 @@ public void awaitMetadataUpdate() {
* until it has completed).
*/
public void ensureFreshMetadata() {
- if (this.metadata.timeToNextUpdate(time.milliseconds()) == 0)
+ if (this.metadata.updateRequested() || this.metadata.timeToNextUpdate(time.milliseconds()) == 0)
awaitMetadataUpdate();
}
@@ -196,10 +194,11 @@ public void poll(long timeout) {
/**
* Poll for network IO and return immediately. This will not trigger wakeups,
* nor will it execute any delayed tasks.
+ * @param executeDelayedTasks Whether to allow delayed task execution (true allows)
*/
- public void quickPoll() {
+ public void quickPoll(boolean executeDelayedTasks) {
disableWakeups();
- poll(0, time.milliseconds(), false);
+ poll(0, time.milliseconds(), executeDelayedTasks);
enableWakeups();
}
@@ -226,8 +225,8 @@ private void poll(long timeout, long now, boolean executeDelayedTasks) {
// cleared or a connect finished in the poll
trySend(now);
- // fail all requests that couldn't be sent
- failUnsentRequests();
+ // fail requests that couldn't be sent if they have expired
+ failExpiredRequests(now);
}
/**
@@ -273,29 +272,48 @@ private void checkDisconnects(long now) {
Map.Entry> requestEntry = iterator.next();
Node node = requestEntry.getKey();
if (client.connectionFailed(node)) {
+ // Remove entry before invoking request callback to avoid callbacks handling
+ // coordinator failures traversing the unsent list again.
+ iterator.remove();
for (ClientRequest request : requestEntry.getValue()) {
RequestFutureCompletionHandler handler =
(RequestFutureCompletionHandler) request.callback();
handler.onComplete(new ClientResponse(request, now, true, null));
}
- iterator.remove();
}
}
}
- private void failUnsentRequests() {
- // clear all unsent requests and fail their corresponding futures
- for (Map.Entry> requestEntry: unsent.entrySet()) {
- Iterator iterator = requestEntry.getValue().iterator();
- while (iterator.hasNext()) {
- ClientRequest request = iterator.next();
- RequestFutureCompletionHandler handler =
- (RequestFutureCompletionHandler) request.callback();
- handler.raise(SendFailedException.INSTANCE);
+ private void failExpiredRequests(long now) {
+ // clear all expired unsent requests and fail their corresponding futures
+ Iterator>> iterator = unsent.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry> requestEntry = iterator.next();
+ Iterator requestIterator = requestEntry.getValue().iterator();
+ while (requestIterator.hasNext()) {
+ ClientRequest request = requestIterator.next();
+ if (request.createdTimeMs() < now - unsentExpiryMs) {
+ RequestFutureCompletionHandler handler =
+ (RequestFutureCompletionHandler) request.callback();
+ handler.raise(new TimeoutException("Failed to send request after " + unsentExpiryMs + " ms."));
+ requestIterator.remove();
+ } else
+ break;
+ }
+ if (requestEntry.getValue().isEmpty())
iterator.remove();
+ }
+ }
+
+ protected void failUnsentRequests(Node node, RuntimeException e) {
+ // clear unsent requests to node and fail their corresponding futures
+ List unsentRequests = unsent.remove(node);
+ if (unsentRequests != null) {
+ for (ClientRequest request : unsentRequests) {
+ RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
+ handler.raise(e);
}
}
- unsent.clear();
}
private boolean trySend(long now) {
@@ -319,7 +337,6 @@ private boolean trySend(long now) {
private void clientPoll(long timeout, long now) {
client.poll(timeout, now);
if (wakeupsEnabled && wakeup.get()) {
- failUnsentRequests();
wakeup.set(false);
throw new WakeupException();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index 7a1a72090a3a3..f6d338747668b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -61,6 +61,7 @@
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
@@ -182,25 +183,26 @@ public void updateFetchPositions(Set partitions) {
* @return The map of topics with their partition information
*/
public Map> getAllTopicMetadata(long timeout) {
- return getTopicMetadata(null, timeout);
+ return getTopicMetadata(MetadataRequest.allTopics(), timeout);
}
/**
* Get metadata for all topics present in Kafka cluster
*
- * @param topics The list of topics to fetch or null to fetch all
+ * @param request The MetadataRequest to send
* @param timeout time for which getting topic metadata is attempted
* @return The map of topics with their partition information
*/
- public Map> getTopicMetadata(List topics, long timeout) {
- if (topics != null && topics.isEmpty())
+ public Map> getTopicMetadata(MetadataRequest request, long timeout) {
+ // Save the round trip if no topics are requested.
+ if (!request.isAllTopics() && request.topics().isEmpty())
return Collections.emptyMap();
long start = time.milliseconds();
long remaining = timeout;
do {
- RequestFuture future = sendMetadataRequest(topics);
+ RequestFuture future = sendMetadataRequest(request);
client.poll(future, remaining);
if (future.failed() && !future.isRetriable())
@@ -215,13 +217,14 @@ public Map> getTopicMetadata(List topics, lo
throw new TopicAuthorizationException(unauthorizedTopics);
boolean shouldRetry = false;
- if (!response.errors().isEmpty()) {
+ Map errors = response.errors();
+ if (!errors.isEmpty()) {
// if there were errors, we need to check whether they were fatal or whether
// we should just retry
- log.debug("Topic metadata fetch included errors: {}", response.errors());
+ log.debug("Topic metadata fetch included errors: {}", errors);
- for (Map.Entry errorEntry : response.errors().entrySet()) {
+ for (Map.Entry errorEntry : errors.entrySet()) {
String topic = errorEntry.getKey();
Errors error = errorEntry.getValue();
@@ -264,14 +267,12 @@ else if (error.exception() instanceof RetriableException)
* Send Metadata Request to least loaded node in Kafka cluster asynchronously
* @return A future that indicates result of sent metadata request
*/
- private RequestFuture sendMetadataRequest(List topics) {
- if (topics == null)
- topics = Collections.emptyList();
+ private RequestFuture sendMetadataRequest(MetadataRequest request) {
final Node node = client.leastLoadedNode();
if (node == null)
return RequestFuture.noBrokersAvailable();
else
- return client.send(node, ApiKeys.METADATA, new MetadataRequest(topics));
+ return client.send(node, ApiKeys.METADATA, request);
}
/**
@@ -290,7 +291,7 @@ else if (strategy == OffsetResetStrategy.LATEST)
else
throw new NoOffsetForPartitionException(partition);
- log.debug("Resetting offset for partition {} to {} offset.", partition, strategy.name().toLowerCase());
+ log.debug("Resetting offset for partition {} to {} offset.", partition, strategy.name().toLowerCase(Locale.ROOT));
long offset = listOffset(partition, timestamp);
// we might lose the assignment while fetching the offset, so check it is still active
@@ -501,12 +502,12 @@ private void handleListOffsetResponse(TopicPartition topicPartition,
future.complete(offset);
} else if (errorCode == Errors.NOT_LEADER_FOR_PARTITION.code()
|| errorCode == Errors.UNKNOWN_TOPIC_OR_PARTITION.code()) {
- log.warn("Attempt to fetch offsets for partition {} failed due to obsolete leadership information, retrying.",
+ log.debug("Attempt to fetch offsets for partition {} failed due to obsolete leadership information, retrying.",
topicPartition);
future.raise(Errors.forCode(errorCode));
} else {
- log.error("Attempt to fetch offsets for partition {} failed due to: {}",
- topicPartition, Errors.forCode(errorCode).exception().getMessage());
+ log.warn("Attempt to fetch offsets for partition {} failed due to: {}",
+ topicPartition, Errors.forCode(errorCode).message());
future.raise(new StaleMetadataException());
}
}
@@ -652,8 +653,8 @@ private ConsumerRecord parseRecord(TopicPartition partition, LogEntry logE
return new ConsumerRecord<>(partition.topic(), partition.partition(), offset,
timestamp, timestampType, logEntry.record().checksum(),
- keyByteArray == null ? -1 : keyByteArray.length,
- valueByteArray == null ? -1 : valueByteArray.length,
+ keyByteArray == null ? ConsumerRecord.NULL_SIZE : keyByteArray.length,
+ valueByteArray == null ? ConsumerRecord.NULL_SIZE : valueByteArray.length,
key, value);
} catch (KafkaException e) {
throw e;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestFuture.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestFuture.java
index 7be99bd2d7097..71c16faf95275 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestFuture.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestFuture.java
@@ -42,7 +42,7 @@ public class RequestFuture {
private boolean isDone = false;
private T value;
private RuntimeException exception;
- private List> listeners = new ArrayList>();
+ private List> listeners = new ArrayList<>();
/**
@@ -129,12 +129,12 @@ public void raise(Errors error) {
}
private void fireSuccess() {
- for (RequestFutureListener listener: listeners)
+ for (RequestFutureListener listener : listeners)
listener.onSuccess(value);
}
private void fireFailure() {
- for (RequestFutureListener listener: listeners)
+ for (RequestFutureListener listener : listeners)
listener.onFailure(exception);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
index af263577ac279..e72a476ee5c52 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
@@ -21,7 +21,6 @@
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
@@ -89,7 +88,7 @@ public SubscriptionState(OffsetResetStrategy defaultResetStrategy) {
this.subscribedPattern = null;
}
- public void subscribe(List topics, ConsumerRebalanceListener listener) {
+ public void subscribe(Collection topics, ConsumerRebalanceListener listener) {
if (listener == null)
throw new IllegalArgumentException("RebalanceListener cannot be null");
@@ -101,7 +100,7 @@ public void subscribe(List topics, ConsumerRebalanceListener listener) {
changeSubscription(topics);
}
- public void changeSubscription(List topicsToSubscribe) {
+ public void changeSubscription(Collection topicsToSubscribe) {
if (!this.subscription.equals(new HashSet<>(topicsToSubscribe))) {
this.subscription.clear();
this.subscription.addAll(topicsToSubscribe);
@@ -415,4 +414,4 @@ private boolean isFetchable() {
}
-}
\ No newline at end of file
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index a066512c9d51a..5b05272ab2213 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -73,7 +73,7 @@
*
* {@code
* Properties props = new Properties();
- * props.put("bootstrap.servers", "localhost:4242");
+ * props.put("bootstrap.servers", "localhost:9092");
* props.put("acks", "all");
* props.put("retries", 0);
* props.put("batch.size", 16384);
@@ -202,7 +202,7 @@ public KafkaProducer(Properties properties, Serializer keySerializer, Seriali
keySerializer, valueSerializer);
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "deprecation"})
private KafkaProducer(ProducerConfig config, Serializer keySerializer, Serializer valueSerializer) {
try {
log.trace("Starting the Kafka producer");
@@ -287,6 +287,7 @@ private KafkaProducer(ProducerConfig config, Serializer keySerializer, Serial
this.sender = new Sender(client,
this.metadata,
this.accumulator,
+ config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION) == 1,
config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
(short) parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),
config.getInt(ProducerConfig.RETRIES_CONFIG),
@@ -337,7 +338,7 @@ private KafkaProducer(ProducerConfig config, Serializer keySerializer, Serial
private static int parseAcks(String acksString) {
try {
- return acksString.trim().toLowerCase().equals("all") ? -1 : Integer.parseInt(acksString.trim());
+ return acksString.trim().equalsIgnoreCase("all") ? -1 : Integer.parseInt(acksString.trim());
} catch (NumberFormatException e) {
throw new ConfigException("Invalid configuration value for 'acks': " + acksString);
}
@@ -426,9 +427,7 @@ public Future send(ProducerRecord record) {
public Future send(ProducerRecord record, Callback callback) {
// intercept the record, which can be potentially modified; this method does not throw exceptions
ProducerRecord interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
- // producer callback will make sure to call both 'callback' and interceptor callback
- Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors);
- return doSend(interceptedRecord, interceptCallback);
+ return doSend(interceptedRecord, callback);
}
/**
@@ -436,6 +435,7 @@ public Future send(ProducerRecord record, Callback callbac
* See {@link #send(ProducerRecord, Callback)} for details.
*/
private Future doSend(ProducerRecord record, Callback callback) {
+ TopicPartition tp = null;
try {
// first make sure the metadata for the topic is available
long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs);
@@ -459,10 +459,12 @@ private Future doSend(ProducerRecord record, Callback call
int partition = partition(record, serializedKey, serializedValue, metadata.fetch());
int serializedSize = Records.LOG_OVERHEAD + Record.recordSize(serializedKey, serializedValue);
ensureValidRecordSize(serializedSize);
- TopicPartition tp = new TopicPartition(record.topic(), partition);
+ tp = new TopicPartition(record.topic(), partition);
long timestamp = record.timestamp() == null ? time.milliseconds() : record.timestamp();
log.trace("Sending record {} with callback {} to topic {} partition {}", record, callback, record.topic(), partition);
- RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, callback, remainingWaitMs);
+ // producer callback will make sure to call both 'callback' and interceptor callback
+ Callback interceptCallback = this.interceptors == null ? callback : new InterceptorCallback<>(callback, this.interceptors, tp);
+ RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey, serializedValue, interceptCallback, remainingWaitMs);
if (result.batchIsFull || result.newBatchCreated) {
log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), partition);
this.sender.wakeup();
@@ -476,27 +478,29 @@ private Future doSend(ProducerRecord record, Callback call
if (callback != null)
callback.onCompletion(null, e);
this.errors.record();
+ if (this.interceptors != null)
+ this.interceptors.onSendError(record, tp, e);
return new FutureFailure(e);
} catch (InterruptedException e) {
this.errors.record();
if (this.interceptors != null)
- this.interceptors.onAcknowledgement(null, e);
+ this.interceptors.onSendError(record, tp, e);
throw new InterruptException(e);
} catch (BufferExhaustedException e) {
this.errors.record();
this.metrics.sensor("buffer-exhausted-records").record();
if (this.interceptors != null)
- this.interceptors.onAcknowledgement(null, e);
+ this.interceptors.onSendError(record, tp, e);
throw e;
} catch (KafkaException e) {
this.errors.record();
if (this.interceptors != null)
- this.interceptors.onAcknowledgement(null, e);
+ this.interceptors.onSendError(record, tp, e);
throw e;
} catch (Exception e) {
// we notify interceptor about all exceptions, since onSend is called before anything else in this method
if (this.interceptors != null)
- this.interceptors.onAcknowledgement(null, e);
+ this.interceptors.onSendError(record, tp, e);
throw e;
}
}
@@ -707,13 +711,11 @@ private int partition(ProducerRecord record, byte[] serializedKey , byte[]
Integer partition = record.partition();
if (partition != null) {
List partitions = cluster.partitionsForTopic(record.topic());
- int numPartitions = partitions.size();
+ int lastPartition = partitions.size() - 1;
// they have given us a partition, use it
- if (partition < 0 || partition >= numPartitions)
- throw new IllegalArgumentException("Invalid partition given with record: " + partition
- + " is not in the range [0..."
- + numPartitions
- + "].");
+ if (partition < 0 || partition > lastPartition) {
+ throw new IllegalArgumentException(String.format("Invalid partition given with record: %d is not in the range [0...%d].", partition, lastPartition));
+ }
return partition;
}
return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue,
@@ -762,15 +764,24 @@ public boolean isDone() {
private static class InterceptorCallback implements Callback {
private final Callback userCallback;
private final ProducerInterceptors interceptors;
+ private final TopicPartition tp;
- public InterceptorCallback(Callback userCallback, ProducerInterceptors interceptors) {
+ public InterceptorCallback(Callback userCallback, ProducerInterceptors interceptors,
+ TopicPartition tp) {
this.userCallback = userCallback;
this.interceptors = interceptors;
+ this.tp = tp;
}
public void onCompletion(RecordMetadata metadata, Exception exception) {
- if (this.interceptors != null)
- this.interceptors.onAcknowledgement(metadata, exception);
+ if (this.interceptors != null) {
+ if (metadata == null) {
+ this.interceptors.onAcknowledgement(new RecordMetadata(tp, -1, -1, Record.NO_TIMESTAMP, -1, -1, -1),
+ exception);
+ } else {
+ this.interceptors.onAcknowledgement(metadata, exception);
+ }
+ }
if (this.userCallback != null)
this.userCallback.onCompletion(metadata, exception);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
index ee2b142a70fde..5b7a296a903b1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
@@ -59,7 +59,8 @@ public class ProducerConfig extends AbstractConfig {
/** batch.size */
public static final String BATCH_SIZE_CONFIG = "batch.size";
- private static final String BATCH_SIZE_DOC = "The producer will attempt to batch records together into fewer requests whenever multiple records are being sent" + " to the same partition. This helps performance on both the client and the server. This configuration controls the "
+ private static final String BATCH_SIZE_DOC = "The producer will attempt to batch records together into fewer requests whenever multiple records are being sent"
+ + " to the same partition. This helps performance on both the client and the server. This configuration controls the "
+ "default batch size in bytes. "
+ ""
+ "No attempt will be made to batch records larger than this size. "
@@ -70,15 +71,6 @@ public class ProducerConfig extends AbstractConfig {
+ "batching entirely). A very large batch size may use memory a bit more wastefully as we will always allocate a "
+ "buffer of the specified batch size in anticipation of additional records.";
- /** buffer.memory */
- public static final String BUFFER_MEMORY_CONFIG = "buffer.memory";
- private static final String BUFFER_MEMORY_DOC = "The total bytes of memory the producer can use to buffer records waiting to be sent to the server. If records are " + "sent faster than they can be delivered to the server the producer will either block or throw an exception based "
- + "on the preference specified by block.on.buffer.full. "
- + "
"
- + "This setting should correspond roughly to the total memory the producer will use, but is not a hard bound since "
- + "not all memory the producer uses is used for buffering. Some additional memory will be used for compression (if "
- + "compression is enabled) as well as for maintaining in-flight requests.";
-
/** acks */
public static final String ACKS_CONFIG = "acks";
private static final String ACKS_DOC = "The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the "
@@ -103,20 +95,22 @@ public class ProducerConfig extends AbstractConfig {
*/
@Deprecated
public static final String TIMEOUT_CONFIG = "timeout.ms";
- private static final String TIMEOUT_DOC = "The configuration controls the maximum amount of time the server will wait for acknowledgments from followers to " + "meet the acknowledgment requirements the producer has specified with the acks configuration. If the "
+ private static final String TIMEOUT_DOC = "The configuration controls the maximum amount of time the server will wait for acknowledgments from followers to "
+ + "meet the acknowledgment requirements the producer has specified with the acks configuration. If the "
+ "requested number of acknowledgments are not met when the timeout elapses an error will be returned. This timeout "
+ "is measured on the server side and does not include the network latency of the request.";
/** linger.ms */
public static final String LINGER_MS_CONFIG = "linger.ms";
- private static final String LINGER_MS_DOC = "The producer groups together any records that arrive in between request transmissions into a single batched request. " + "Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to "
+ private static final String LINGER_MS_DOC = "The producer groups together any records that arrive in between request transmissions into a single batched request. "
+ + "Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to "
+ "reduce the number of requests even under moderate load. This setting accomplishes this by adding a small amount "
+ "of artificial delay—that is, rather than immediately sending out a record the producer will wait for up to "
+ "the given delay to allow other records to be sent so that the sends can be batched together. This can be thought "
+ "of as analogous to Nagle's algorithm in TCP. This setting gives the upper bound on the delay for batching: once "
- + "we get batch.size worth of records for a partition it will be sent immediately regardless of this "
+ + "we get " + BATCH_SIZE_CONFIG + " worth of records for a partition it will be sent immediately regardless of this "
+ "setting, however if we have fewer than this many bytes accumulated for this partition we will 'linger' for the "
- + "specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting linger.ms=5, "
+ + "specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting " + LINGER_MS_CONFIG + "=5, "
+ "for example, would have the effect of reducing the number of requests sent but would add up to 5ms of latency to records sent in the absense of load.";
/** client.id */
@@ -130,24 +124,47 @@ public class ProducerConfig extends AbstractConfig {
/** max.request.size */
public static final String MAX_REQUEST_SIZE_CONFIG = "max.request.size";
- private static final String MAX_REQUEST_SIZE_DOC = "The maximum size of a request in bytes. This is also effectively a cap on the maximum record size. Note that the server " + "has its own cap on record size which may be different from this. This setting will limit the number of record "
+ private static final String MAX_REQUEST_SIZE_DOC = "The maximum size of a request in bytes. This is also effectively a cap on the maximum record size. Note that the server "
+ + "has its own cap on record size which may be different from this. This setting will limit the number of record "
+ "batches the producer will send in a single request to avoid sending huge requests.";
/** reconnect.backoff.ms */
public static final String RECONNECT_BACKOFF_MS_CONFIG = CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG;
+ /** max.block.ms */
+ public static final String MAX_BLOCK_MS_CONFIG = "max.block.ms";
+ private static final String MAX_BLOCK_MS_DOC = "The configuration controls how long KafkaProducer.send() and KafkaProducer.partitionsFor() will block."
+ + "These methods can be blocked either because the buffer is full or metadata unavailable."
+ + "Blocking in the user-supplied serializers or partitioner will not be counted against this timeout.";
+
/** block.on.buffer.full */
/**
* @deprecated This config will be removed in a future release. Also, the {@link #METADATA_FETCH_TIMEOUT_CONFIG} is no longer honored when this property is set to true.
*/
@Deprecated
public static final String BLOCK_ON_BUFFER_FULL_CONFIG = "block.on.buffer.full";
- private static final String BLOCK_ON_BUFFER_FULL_DOC = "When our memory buffer is exhausted we must either stop accepting new records (block) or throw errors. By default " + "this setting is true and we block, however in some scenarios blocking is not desirable and it is better to "
- + "immediately give an error. Setting this to false will accomplish that: the producer will throw a BufferExhaustedException if a record is sent and the buffer space is full.";
+ private static final String BLOCK_ON_BUFFER_FULL_DOC = "When our memory buffer is exhausted we must either stop accepting new records (block) or throw errors. "
+ + "By default this setting is false and the producer will throw a BufferExhaustedException if a record is sent and the buffer space is full. "
+ + "However in some scenarios getting an error is not desirable and it is better to block. Setting this to true will accomplish that."
+ + "If this property is set to true, parameter " + METADATA_FETCH_TIMEOUT_CONFIG + " is not longer honored."
+ + "
"
+ + "This parameter is deprecated and will be removed in a future release. "
+ + "Parameter " + MAX_BLOCK_MS_CONFIG + " should be used instead.";
+
+ /** buffer.memory */
+ public static final String BUFFER_MEMORY_CONFIG = "buffer.memory";
+ private static final String BUFFER_MEMORY_DOC = "The total bytes of memory the producer can use to buffer records waiting to be sent to the server. If records are "
+ + "sent faster than they can be delivered to the server the producer will either block or throw an exception based "
+ + "on the preference specified by " + BLOCK_ON_BUFFER_FULL_CONFIG + ". "
+ + "
"
+ + "This setting should correspond roughly to the total memory the producer will use, but is not a hard bound since "
+ + "not all memory the producer uses is used for buffering. Some additional memory will be used for compression (if "
+ + "compression is enabled) as well as for maintaining in-flight requests.";
/** retries */
public static final String RETRIES_CONFIG = "retries";
- private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error." + " Note that this retry is no different than if the client resent the record upon receiving the "
+ private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error."
+ + " Note that this retry is no different than if the client resent the record upon receiving the "
+ "error. Allowing retries will potentially change the ordering of records because if two records are "
+ "sent to a single partition, and the first fails and is retried but the second succeeds, then the second record "
+ "may appear first.";
@@ -157,7 +174,8 @@ public class ProducerConfig extends AbstractConfig {
/** compression.type */
public static final String COMPRESSION_TYPE_CONFIG = "compression.type";
- private static final String COMPRESSION_TYPE_DOC = "The compression type for all data generated by the producer. The default is none (i.e. no compression). Valid " + " values are none, gzip, snappy, or lz4. "
+ private static final String COMPRESSION_TYPE_DOC = "The compression type for all data generated by the producer. The default is none (i.e. no compression). Valid "
+ + " values are none, gzip, snappy, or lz4. "
+ "Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression).";
/** metrics.sample.window.ms */
@@ -190,12 +208,6 @@ public class ProducerConfig extends AbstractConfig {
public static final String PARTITIONER_CLASS_CONFIG = "partitioner.class";
private static final String PARTITIONER_CLASS_DOC = "Partitioner class that implements the Partitioner interface.";
- /** max.block.ms */
- public static final String MAX_BLOCK_MS_CONFIG = "max.block.ms";
- private static final String MAX_BLOCK_MS_DOC = "The configuration controls how long {@link KafkaProducer#send()} and {@link KafkaProducer#partitionsFor} will block."
- + "These methods can be blocked either because the buffer is full or metadata unavailable."
- + "Blocking in the user-supplied serializers or partitioner will not be counted against this timeout.";
-
/** request.timeout.ms */
public static final String REQUEST_TIMEOUT_MS_CONFIG = CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG;
private static final String REQUEST_TIMEOUT_MS_DOC = CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC;
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java
index aa18fdc736c2b..e835a69501b2b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerInterceptor.java
@@ -76,7 +76,12 @@ public interface ProducerInterceptor extends Configurable {
* This method will generally execute in the background I/O thread, so the implementation should be reasonably fast.
* Otherwise, sending of messages from other threads could be delayed.
*
- * @param metadata The metadata for the record that was sent (i.e. the partition and offset). Null if an error occurred.
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will contain only valid topic and maybe
+ * partition. If partition is not given in ProducerRecord and an error occurs
+ * before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
+ * The metadata may be null if the client passed null record to
+ * {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
public void onAcknowledgement(RecordMetadata metadata, Exception exception);
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java b/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
index c60a53d916209..00a4374ef5543 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
@@ -17,12 +17,18 @@
package org.apache.kafka.clients.producer;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.record.Record;
/**
* The metadata for a record that has been acknowledged by the server
*/
public final class RecordMetadata {
+ /**
+ * Partition value for record without partition assigned
+ */
+ public static final int UNKNOWN_PARTITION = -1;
+
private final long offset;
// The timestamp of the message.
// If LogAppendTime is used for the topic, the timestamp will be the timestamp returned by the broker.
@@ -46,6 +52,11 @@ private RecordMetadata(TopicPartition topicPartition, long offset, long timestam
this.topicPartition = topicPartition;
}
+ @Deprecated
+ public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset) {
+ this(topicPartition, baseOffset, relativeOffset, Record.NO_TIMESTAMP, -1, -1, -1);
+ }
+
public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset,
long timestamp, long checksum, int serializedKeySize, int serializedValueSize) {
// ignore the relativeOffset if the base offset is -1,
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ErrorLoggingCallback.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ErrorLoggingCallback.java
index 747e29fadbcf6..18088c1c04291 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ErrorLoggingCallback.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ErrorLoggingCallback.java
@@ -44,8 +44,8 @@ public void onCompletion(RecordMetadata metadata, Exception e) {
logAsString ? new String(key) : key.length + " bytes";
String valueString = (valueLength == -1) ? "null" :
logAsString ? new String(value) : valueLength + " bytes";
- log.error("Error when sending message to topic {} with key: {}, value: {} with error: {}",
- topic, keyString, valueString, e.getMessage());
+ log.error("Error when sending message to topic {} with key: {}, value: {} with error:",
+ topic, keyString, valueString, e);
}
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java
index 9343a2e188988..8466d3a85752d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerInterceptors.java
@@ -20,6 +20,8 @@
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.record.Record;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,7 +78,8 @@ public ProducerRecord onSend(ProducerRecord record) {
*
* This method does not throw exceptions. Exceptions thrown by any of interceptor methods are caught and ignored.
*
- * @param metadata The metadata for the record that was sent (i.e. the partition and offset). Null if an error occurred.
+ * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
+ * If an error occurred, metadata will only contain valid topic and maybe partition.
* @param exception The exception thrown during processing of this record. Null if no error occurred.
*/
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
@@ -90,6 +93,36 @@ public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
}
}
+ /**
+ * This method is called when sending the record fails in {@link ProducerInterceptor#onSend
+ * (ProducerRecord)} method. This method calls {@link ProducerInterceptor#onAcknowledgement(RecordMetadata, Exception)}
+ * method for each interceptor
+ *
+ * @param record The record from client
+ * @param interceptTopicPartition The topic/partition for the record if an error occurred
+ * after partition gets assigned; the topic part of interceptTopicPartition is the same as in record.
+ * @param exception The exception thrown during processing of this record.
+ */
+ public void onSendError(ProducerRecord record, TopicPartition interceptTopicPartition, Exception exception) {
+ for (ProducerInterceptor interceptor : this.interceptors) {
+ try {
+ if (record == null && interceptTopicPartition == null) {
+ interceptor.onAcknowledgement(null, exception);
+ } else {
+ if (interceptTopicPartition == null) {
+ interceptTopicPartition = new TopicPartition(record.topic(),
+ record.partition() == null ? RecordMetadata.UNKNOWN_PARTITION : record.partition());
+ }
+ interceptor.onAcknowledgement(new RecordMetadata(interceptTopicPartition, -1, -1, Record.NO_TIMESTAMP, -1, -1, -1),
+ exception);
+ }
+ } catch (Exception e) {
+ // do not propagate interceptor exceptions, just log
+ log.warn("Error executing interceptor onAcknowledgement callback", e);
+ }
+ }
+ }
+
/**
* Closes every interceptor in a container.
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
index f1414f01472e2..1766609ace026 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
@@ -13,6 +13,7 @@
package org.apache.kafka.clients.producer.internals;
import java.util.Iterator;
+
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.MetricName;
@@ -59,7 +60,6 @@ public final class RecordAccumulator {
private static final Logger log = LoggerFactory.getLogger(RecordAccumulator.class);
private volatile boolean closed;
- private int drainIndex;
private final AtomicInteger flushesInProgress;
private final AtomicInteger appendsInProgress;
private final int batchSize;
@@ -70,6 +70,9 @@ public final class RecordAccumulator {
private final Time time;
private final ConcurrentMap> batches;
private final IncompleteRecordBatches incomplete;
+ // The following variables are only accessed by the sender thread, so we don't need to protect them.
+ private final Set muted;
+ private int drainIndex;
/**
@@ -105,6 +108,7 @@ public RecordAccumulator(int batchSize,
String metricGrpName = "producer-metrics";
this.free = new BufferPool(totalSize, batchSize, metrics, time, metricGrpName);
this.incomplete = new IncompleteRecordBatches();
+ this.muted = new HashSet();
this.time = time;
registerMetrics(metrics, metricGrpName);
}
@@ -163,7 +167,7 @@ public RecordAppendResult append(TopicPartition tp,
appendsInProgress.incrementAndGet();
try {
// check if we have an in-progress batch
- Deque dq = dequeFor(tp);
+ Deque dq = getOrCreateDeque(tp);
synchronized (dq) {
if (closed)
throw new IllegalStateException("Cannot send after the producer is closed.");
@@ -209,25 +213,32 @@ public RecordAppendResult append(TopicPartition tp,
* Abort the batches that have been sitting in RecordAccumulator for more than the configured requestTimeout
* due to metadata being unavailable
*/
- public List abortExpiredBatches(int requestTimeout, Cluster cluster, long now) {
+ public List abortExpiredBatches(int requestTimeout, long now) {
List expiredBatches = new ArrayList();
int count = 0;
for (Map.Entry> entry : this.batches.entrySet()) {
- TopicPartition topicAndPartition = entry.getKey();
Deque dq = entry.getValue();
- synchronized (dq) {
- // iterate over the batches and expire them if they have stayed in accumulator for more than requestTimeOut
- Iterator batchIterator = dq.iterator();
- while (batchIterator.hasNext()) {
- RecordBatch batch = batchIterator.next();
- // check if the batch is expired
- if (batch.maybeExpire(requestTimeout, now, this.lingerMs)) {
- expiredBatches.add(batch);
- count++;
- batchIterator.remove();
- deallocate(batch);
- } else {
- if (!batch.inRetry()) {
+ TopicPartition tp = entry.getKey();
+ // We only check if the batch should be expired if the partition does not have a batch in flight.
+ // This is to prevent later batches from being expired while an earlier batch is still in progress.
+ // Note that `muted` is only ever populated if `max.in.flight.request.per.connection=1` so this protection
+ // is only active in this case. Otherwise the expiration order is not guaranteed.
+ if (!muted.contains(tp)) {
+ synchronized (dq) {
+ // iterate over the batches and expire them if they have been in the accumulator for more than requestTimeOut
+ RecordBatch lastBatch = dq.peekLast();
+ Iterator batchIterator = dq.iterator();
+ while (batchIterator.hasNext()) {
+ RecordBatch batch = batchIterator.next();
+ boolean isFull = batch != lastBatch || batch.records.isFull();
+ // check if the batch is expired
+ if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) {
+ expiredBatches.add(batch);
+ count++;
+ batchIterator.remove();
+ deallocate(batch);
+ } else {
+ // Stop at the first batch that has not expired.
break;
}
}
@@ -248,7 +259,7 @@ public void reenqueue(RecordBatch batch, long now) {
batch.lastAttemptMs = now;
batch.lastAppendTime = now;
batch.setRetry();
- Deque deque = dequeFor(batch.topicPartition);
+ Deque deque = getOrCreateDeque(batch.topicPartition);
synchronized (deque) {
deque.addFirst(batch);
}
@@ -259,14 +270,20 @@ public void reenqueue(RecordBatch batch, long now) {
* partition will be ready; Also return the flag for whether there are any unknown leaders for the accumulated
* partition batches.
*
- * A destination node is ready to send data if ANY one of its partition is not backing off the send and ANY of the
- * following are true :
+ * A destination node is ready to send data if:
*
- * - The record set is full
- *
- The record set has sat in the accumulator for at least lingerMs milliseconds
- *
- The accumulator is out of memory and threads are blocking waiting for data (in this case all partitions are
- * immediately considered ready).
- *
- The accumulator has been closed
+ *
- There is at least one partition that is not backing off its send
+ *
- and those partitions are not muted (to prevent reordering if
+ * {@value org.apache.kafka.clients.producer.ProducerConfig#MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION}
+ * is set to one)
+ * - and any of the following are true
+ *
+ * - The record set is full
+ * - The record set has sat in the accumulator for at least lingerMs milliseconds
+ * - The accumulator is out of memory and threads are blocking waiting for data (in this case all partitions
+ * are immediately considered ready).
+ * - The accumulator has been closed
+ *
*
*/
public ReadyCheckResult ready(Cluster cluster, long nowMs) {
@@ -282,7 +299,7 @@ public ReadyCheckResult ready(Cluster cluster, long nowMs) {
Node leader = cluster.leaderFor(part);
if (leader == null) {
unknownLeadersExist = true;
- } else if (!readyNodes.contains(leader)) {
+ } else if (!readyNodes.contains(leader) && !muted.contains(part)) {
synchronized (deque) {
RecordBatch batch = deque.peekFirst();
if (batch != null) {
@@ -333,7 +350,10 @@ public boolean hasUnsent() {
* @param now The current unix time in milliseconds
* @return A list of {@link RecordBatch} for each node specified with total size less than the requested maxSize.
*/
- public Map> drain(Cluster cluster, Set nodes, int maxSize, long now) {
+ public Map> drain(Cluster cluster,
+ Set nodes,
+ int maxSize,
+ long now) {
if (nodes.isEmpty())
return Collections.emptyMap();
@@ -346,25 +366,29 @@ public Map> drain(Cluster cluster, Set nodes, i
int start = drainIndex = drainIndex % parts.size();
do {
PartitionInfo part = parts.get(drainIndex);
- Deque deque = dequeFor(new TopicPartition(part.topic(), part.partition()));
- if (deque != null) {
- synchronized (deque) {
- RecordBatch first = deque.peekFirst();
- if (first != null) {
- boolean backoff = first.attempts > 0 && first.lastAttemptMs + retryBackoffMs > now;
- // Only drain the batch if it is not during backoff period.
- if (!backoff) {
- if (size + first.records.sizeInBytes() > maxSize && !ready.isEmpty()) {
- // there is a rare case that a single batch size is larger than the request size due
- // to compression; in this case we will still eventually send this batch in a single
- // request
- break;
- } else {
- RecordBatch batch = deque.pollFirst();
- batch.records.close();
- size += batch.records.sizeInBytes();
- ready.add(batch);
- batch.drainedMs = now;
+ TopicPartition tp = new TopicPartition(part.topic(), part.partition());
+ // Only proceed if the partition has no in-flight batches.
+ if (!muted.contains(tp)) {
+ Deque deque = getDeque(new TopicPartition(part.topic(), part.partition()));
+ if (deque != null) {
+ synchronized (deque) {
+ RecordBatch first = deque.peekFirst();
+ if (first != null) {
+ boolean backoff = first.attempts > 0 && first.lastAttemptMs + retryBackoffMs > now;
+ // Only drain the batch if it is not during backoff period.
+ if (!backoff) {
+ if (size + first.records.sizeInBytes() > maxSize && !ready.isEmpty()) {
+ // there is a rare case that a single batch size is larger than the request size due
+ // to compression; in this case we will still eventually send this batch in a single
+ // request
+ break;
+ } else {
+ RecordBatch batch = deque.pollFirst();
+ batch.records.close();
+ size += batch.records.sizeInBytes();
+ ready.add(batch);
+ batch.drainedMs = now;
+ }
}
}
}
@@ -377,10 +401,14 @@ public Map> drain(Cluster cluster, Set nodes, i
return batches;
}
+ private Deque getDeque(TopicPartition tp) {
+ return batches.get(tp);
+ }
+
/**
* Get the deque for the given topic-partition, creating it if necessary.
*/
- private Deque dequeFor(TopicPartition tp) {
+ private Deque getOrCreateDeque(TopicPartition tp) {
Deque d = this.batches.get(tp);
if (d != null)
return d;
@@ -443,7 +471,7 @@ public void abortIncompleteBatches() {
abortBatches();
} while (appendsInProgress());
// After this point, no thread will append any messages because they will see the close
- // flag set. We need to do the last abort after no thread was appending in case the there was a new
+ // flag set. We need to do the last abort after no thread was appending in case there was a new
// batch appended by the last appending thread.
abortBatches();
this.batches.clear();
@@ -454,7 +482,7 @@ public void abortIncompleteBatches() {
*/
private void abortBatches() {
for (RecordBatch batch : incomplete.all()) {
- Deque dq = dequeFor(batch.topicPartition);
+ Deque dq = getDeque(batch.topicPartition);
// Close the batch before aborting
synchronized (dq) {
batch.records.close();
@@ -465,6 +493,14 @@ private void abortBatches() {
}
}
+ public void mutePartition(TopicPartition tp) {
+ muted.add(tp);
+ }
+
+ public void unmutePartition(TopicPartition tp) {
+ muted.remove(tp);
+ }
+
/**
* Close this accumulator and force all the record buffers to be drained
*/
@@ -532,4 +568,5 @@ public Iterable all() {
}
}
}
+
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java
index 7b5fbbe0736dd..e6cd68fd065e7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java
@@ -134,16 +134,25 @@ public String toString() {
}
/**
- * Expire the batch that is ready but is sitting in accumulator for more than requestTimeout due to metadata being unavailable.
- * We need to explicitly check if the record is full or linger time is met because the accumulator's partition may not be ready
- * if the leader is unavailable.
+ * A batch whose metadata is not available should be expired if one of the following is true:
+ *
+ * - the batch is not in retry AND request timeout has elapsed after it is ready (full or linger.ms has reached).
+ *
- the batch is in retry AND request timeout has elapsed after the backoff period ended.
+ *
*/
- public boolean maybeExpire(int requestTimeout, long now, long lingerMs) {
+ public boolean maybeExpire(int requestTimeoutMs, long retryBackoffMs, long now, long lingerMs, boolean isFull) {
boolean expire = false;
- if ((this.records.isFull() && requestTimeout < (now - this.lastAppendTime)) || requestTimeout < (now - (this.lastAttemptMs + lingerMs))) {
+
+ if (!this.inRetry() && isFull && requestTimeoutMs < (now - this.lastAppendTime))
+ expire = true;
+ else if (!this.inRetry() && requestTimeoutMs < (now - (this.createdMs + lingerMs)))
expire = true;
+ else if (this.inRetry() && requestTimeoutMs < (now - (this.lastAttemptMs + retryBackoffMs)))
+ expire = true;
+
+ if (expire) {
this.records.close();
- this.done(-1L, Record.NO_TIMESTAMP, new TimeoutException("Batch Expired"));
+ this.done(-1L, Record.NO_TIMESTAMP, new TimeoutException("Batch containing " + recordCount + " record(s) expired due to timeout while requesting metadata from brokers for " + topicPartition));
}
return expire;
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
index 9d24d07fc28a6..29077b6bb6b3c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -67,6 +67,9 @@ public class Sender implements Runnable {
/* the metadata for the client */
private final Metadata metadata;
+ /* the flag indicating whether the producer should guarantee the message order on the broker or not. */
+ private final boolean guaranteeMessageOrder;
+
/* the maximum request size to attempt to send to the server */
private final int maxRequestSize;
@@ -97,6 +100,7 @@ public class Sender implements Runnable {
public Sender(KafkaClient client,
Metadata metadata,
RecordAccumulator accumulator,
+ boolean guaranteeMessageOrder,
int maxRequestSize,
short acks,
int retries,
@@ -107,6 +111,7 @@ public Sender(KafkaClient client,
this.client = client;
this.accumulator = accumulator;
this.metadata = metadata;
+ this.guaranteeMessageOrder = guaranteeMessageOrder;
this.maxRequestSize = maxRequestSize;
this.running = true;
this.acks = acks;
@@ -164,7 +169,7 @@ public void run() {
* @param now
* The current POSIX time in milliseconds
*/
- public void run(long now) {
+ void run(long now) {
Cluster cluster = metadata.fetch();
// get the list of partitions with data ready to send
RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);
@@ -189,8 +194,15 @@ public void run(long now) {
result.readyNodes,
this.maxRequestSize,
now);
+ if (guaranteeMessageOrder) {
+ // Mute all the partitions drained
+ for (List batchList : batches.values()) {
+ for (RecordBatch batch : batchList)
+ this.accumulator.mutePartition(batch.topicPartition);
+ }
+ }
- List expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, cluster, now);
+ List expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);
// update sensors
for (RecordBatch expiredBatch : expiredBatches)
this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount);
@@ -304,6 +316,9 @@ private void completeBatch(RecordBatch batch, Errors error, long baseOffset, lon
}
if (error.exception() instanceof InvalidMetadataException)
metadata.requestUpdate();
+ // Unmute the completed partition.
+ if (guaranteeMessageOrder)
+ this.accumulator.unmutePartition(batch.topicPartition);
}
/**
diff --git a/clients/src/main/java/org/apache/kafka/common/BrokerEndPoint.java b/clients/src/main/java/org/apache/kafka/common/BrokerEndPoint.java
deleted file mode 100644
index d5275c419a669..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/BrokerEndPoint.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.common;
-
-import java.io.Serializable;
-
-/**
- * Broker id, host and port
- */
-public final class BrokerEndPoint implements Serializable {
-
- private int hash = 0;
- private final int id;
- private final String host;
- private final int port;
-
- public BrokerEndPoint(int id, String host, int port) {
- this.id = id;
- this.host = host;
- this.port = port;
- }
-
- public int id() {
- return id;
- }
-
- public String host() {
- return host;
- }
-
- public int port() {
- return port;
- }
-
- @Override
- public int hashCode() {
- if (hash != 0)
- return hash;
- final int prime = 31;
- int result = 1;
- result = prime * result + id;
- result = prime * result + ((host == null) ? 0 : host.hashCode());
- result = prime * result + port;
- this.hash = result;
- return result;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj)
- return true;
- if (obj == null)
- return false;
- if (getClass() != obj.getClass())
- return false;
- BrokerEndPoint other = (BrokerEndPoint) obj;
- if (id != other.id)
- return false;
- if (port != other.port)
- return false;
- if (host == null) {
- if (other.host != null)
- return false;
- } else if (!host.equals(other.host))
- return false;
- return true;
- }
-
- @Override
- public String toString() {
- return "[" + id + ", " + host + ":" + port + "]";
- }
-
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/Cluster.java b/clients/src/main/java/org/apache/kafka/common/Cluster.java
index d86e3a4a1b63e..e1bf581b3e5d1 100644
--- a/clients/src/main/java/org/apache/kafka/common/Cluster.java
+++ b/clients/src/main/java/org/apache/kafka/common/Cluster.java
@@ -19,6 +19,7 @@
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -28,6 +29,7 @@
*/
public final class Cluster {
+ private final boolean isBootstrapConfigured;
private final List nodes;
private final Set unauthorizedTopics;
private final Map partitionsByTopicPartition;
@@ -44,13 +46,21 @@ public final class Cluster {
public Cluster(Collection nodes,
Collection partitions,
Set unauthorizedTopics) {
+ this(false, nodes, partitions, unauthorizedTopics);
+ }
+
+ private Cluster(boolean isBootstrapConfigured,
+ Collection nodes,
+ Collection partitions,
+ Set unauthorizedTopics) {
+ this.isBootstrapConfigured = isBootstrapConfigured;
+
// make a randomized, unmodifiable copy of the nodes
List copy = new ArrayList<>(nodes);
Collections.shuffle(copy);
this.nodes = Collections.unmodifiableList(copy);
-
this.nodesById = new HashMap<>();
- for (Node node: nodes)
+ for (Node node : nodes)
this.nodesById.put(node.id(), node);
// index the partitions by topic/partition for quick lookup
@@ -110,47 +120,20 @@ public static Cluster empty() {
* @return A cluster for these hosts/ports
*/
public static Cluster bootstrap(List addresses) {
- List nodes = new ArrayList();
+ List nodes = new ArrayList<>();
int nodeId = -1;
for (InetSocketAddress address : addresses)
- nodes.add(new Node(nodeId--, address.getHostName(), address.getPort()));
- return new Cluster(nodes, new ArrayList(0), Collections.emptySet());
+ nodes.add(new Node(nodeId--, address.getHostString(), address.getPort()));
+ return new Cluster(true, nodes, new ArrayList(0), Collections.emptySet());
}
/**
- * Update the cluster information for specific topic with new partition information
+ * Return a copy of this cluster combined with `partitions`.
*/
- public Cluster update(String topic, Collection partitions) {
-
- // re-index the partitions by topic/partition for quick lookup
- for (PartitionInfo p : partitions)
- this.partitionsByTopicPartition.put(new TopicPartition(p.topic(), p.partition()), p);
-
- // re-index the partitions by topic and node respectively
- this.partitionsByTopic.put(topic, Collections.unmodifiableList(new ArrayList<>(partitions)));
-
- List availablePartitions = new ArrayList<>();
- for (PartitionInfo part : partitions) {
- if (part.leader() != null)
- availablePartitions.add(part);
- }
- this.availablePartitionsByTopic.put(topic, Collections.unmodifiableList(availablePartitions));
-
- HashMap> partsForNode = new HashMap<>();
- for (Node n : this.nodes) {
- partsForNode.put(n.id(), new ArrayList());
- }
- for (PartitionInfo p : partitions) {
- if (p.leader() != null) {
- List psNode = Utils.notNull(partsForNode.get(p.leader().id()));
- psNode.add(p);
- }
- }
-
- for (Map.Entry> entry : partsForNode.entrySet())
- this.partitionsByNode.put(entry.getKey(), Collections.unmodifiableList(entry.getValue()));
-
- return this;
+ public Cluster withPartitions(Map partitions) {
+ Map