> offsets, Exception exception);
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index 76bfe7e91a149..7ec147e9e3ced 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -198,10 +198,7 @@ public class ConsumerConfig extends AbstractConfig {
* fetch.max.wait.ms
*/
public static final String FETCH_MAX_WAIT_MS_CONFIG = "fetch.max.wait.ms";
- private static final String FETCH_MAX_WAIT_MS_DOC = "The maximum amount of time the server will block before " +
- "answering the fetch request there isn't sufficient data to immediately satisfy the requirement given by " +
- "fetch.min.bytes. This config is used only for local log fetch. To tune the remote fetch maximum wait " +
- "time, please refer to 'remote.fetch.max.wait.ms' broker config";
+ private static final String FETCH_MAX_WAIT_MS_DOC = "The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by fetch.min.bytes.";
public static final int DEFAULT_FETCH_MAX_WAIT_MS = 500;
/** metadata.max.age.ms */
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java
index 8884c0393d608..2f43b603fc8ff 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.java
@@ -152,7 +152,7 @@ public interface ConsumerRebalanceListener {
* {@link #onPartitionsRevoked(Collection)} callback before any instance executes its
* {@link #onPartitionsAssigned(Collection)} callback. During exceptional scenarios, partitions may be migrated
* without the old owner being notified (i.e. their {@link #onPartitionsRevoked(Collection)} callback not triggered),
- * and later when the old owner consumer realized this event, the {@link #onPartitionsLost(Collection)} callback
+ * and later when the old owner consumer realized this event, the {@link #onPartitionsLost(Collection)} (Collection)} callback
* will be triggered by the consumer then.
*
* It is common for the assignment callback to use the consumer instance in order to query offsets. It is possible
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java
deleted file mode 100644
index 2fc721a84ff7f..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java
+++ /dev/null
@@ -1,700 +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.clients.consumer;
-
-import org.apache.kafka.clients.KafkaClient;
-import org.apache.kafka.clients.consumer.internals.ConsumerMetadata;
-import org.apache.kafka.clients.consumer.internals.ShareConsumerDelegate;
-import org.apache.kafka.clients.consumer.internals.ShareConsumerDelegateCreator;
-import org.apache.kafka.clients.consumer.internals.SubscriptionState;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.Metric;
-import org.apache.kafka.common.MetricName;
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.annotation.InterfaceStability;
-import org.apache.kafka.common.errors.AuthenticationException;
-import org.apache.kafka.common.errors.AuthorizationException;
-import org.apache.kafka.common.errors.InterruptException;
-import org.apache.kafka.common.errors.InvalidTopicException;
-import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.utils.LogContext;
-import org.apache.kafka.common.utils.Time;
-
-import java.time.Duration;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.Set;
-
-import static org.apache.kafka.common.utils.Utils.propsToMap;
-
-/**
- * A client that consumes records from a Kafka cluster using a share group.
- *
- * This is an early access feature introduced by KIP-932. It is not suitable for production use until it is
- * fully implemented and released.
- *
- *
Cross-Version Compatibility
- * This client can communicate with brokers that are version 4.0.0 or newer. You will receive an
- * {@link org.apache.kafka.common.errors.UnsupportedVersionException} when invoking an API that is not
- * available on the running broker version.
- *
- *
- * Kafka uses the concept of share groups to allow a pool of consumers to cooperate on the work of
- * consuming and processing records. All consumer instances sharing the same {@code group.id} will be part of
- * the same share group.
- *
- * Each consumer in a group can dynamically set the list of topics it wants to subscribe to using the
- * {@link #subscribe(Collection)} method. Kafka will deliver each message in the subscribed topics to one
- * consumer in the share group. Unlike consumer groups, share groups balance the partitions between all
- * members of the share group permitting multiple consumers to consume from the same partitions. This gives
- * more flexible sharing of records than a consumer group, at the expense of record ordering.
- *
- * Membership in a share group is maintained dynamically: if a consumer fails, the partitions assigned to
- * it will be reassigned to other consumers in the same group. Similarly, if a new consumer joins the group,
- * the partition assignment is re-evaluated and partitions can be moved from existing consumers to the new one.
- * This is known as rebalancing the group and is discussed in more detail below.
- * Group rebalancing is also used when new partitions are added to one of the subscribed topics. The group will
- * automatically detect the new partitions through periodic metadata refreshes and assign them to the members of the group.
- *
- * Conceptually, you can think of a share group as a single logical subscriber made up of multiple consumers.
- * In fact, in other messaging systems, a share group is roughly equivalent to a durable shared subscription.
- * You can have multiple share groups and consumer groups independently consuming from the same topics.
- *
- *
- * After subscribing to a set of topics, the consumer will automatically join the group when {@link #poll(Duration)} is
- * invoked. This method 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 records from the partitions it was assigned. Under the covers,
- * the consumer sends periodic heartbeats to the broker. If the consumer crashes or is unable to send heartbeats for
- * the duration of the share group's session time-out, then the consumer will be considered dead and its partitions
- * will be reassigned.
- *
- * It is also possible that the consumer could encounter a "livelock" situation where it is continuing to send heartbeats
- * in the background, but no progress is being made. To prevent the consumer from holding onto its partitions
- * indefinitely in this case, we provide a liveness detection mechanism using the {@code max.poll.interval.ms} setting.
- * If you don't call poll at least as frequently as this, the client will proactively leave the share group.
- * So to stay in the group, you must continue to call poll.
- *
- *
Record Delivery and Acknowledgement
- * When a consumer in a share-group fetches records using {@link #poll(Duration)}, it receives available records from any
- * of the topic-partitions that match its subscriptions. Records are acquired for delivery to this consumer with a
- * time-limited acquisition lock. While a record is acquired, it is not available for another consumer. By default,
- * the lock duration is 30 seconds, but it can also be controlled using the group {@code group.share.record.lock.duration.ms}
- * configuration parameter. The idea is that the lock is automatically released once the lock duration has elapsed, and
- * then the record is available to be given to another consumer. The consumer which holds the lock can deal with it in
- * the following ways:
- *
- * - The consumer can acknowledge successful processing of the record
- * - The consumer can release the record, which makes the record available for another delivery attempt
- * - The consumer can reject the record, which indicates that the record is unprocessable and does not make
- * the record available for another delivery attempt
- * - The consumer can do nothing, in which case the lock is automatically released when the lock duration has elapsed
- *
- * The cluster limits the number of records acquired for consumers for each topic-partition in a share group. Once the limit
- * is reached, fetching records will temporarily yield no further records until the number of acquired records reduces,
- * as naturally happens when the locks time out. This limit is controlled by the broker configuration property
- * {@code group.share.record.lock.partition.limit}. By limiting the duration of the acquisition lock and automatically
- * releasing the locks, the broker ensures delivery progresses even in the presence of consumer failures.
- *
- * The consumer can choose to use implicit or explicit acknowledgement of the records it processes.
- *
If the application calls {@link #acknowledge(ConsumerRecord, AcknowledgeType)} for any record in the batch,
- * it is using explicit acknowledgement. In this case:
- *
- * - The application calls {@link #commitSync()} or {@link #commitAsync()} which commits the acknowledgements to Kafka.
- * If any records in the batch were not acknowledged, they remain acquired and will be presented to the application
- * in response to a future poll.
- * - The application calls {@link #poll(Duration)} without committing first, which commits the acknowledgements to
- * Kafka asynchronously. In this case, no exception is thrown by a failure to commit the acknowledgement.
- * If any records in the batch were not acknowledged, they remain acquired and will be presented to the application
- * in response to a future poll.
- * - The application calls {@link #close()} which attempts to commit any pending acknowledgements and
- * releases any remaining acquired records.
- *
- * If the application does not call {@link #acknowledge(ConsumerRecord, AcknowledgeType)} for any record in the batch,
- * it is using implicit acknowledgement. In this case:
- *
- * - The application calls {@link #commitSync()} or {@link #commitAsync()} which implicitly acknowledges all of
- * the delivered records as processed successfully and commits the acknowledgements to Kafka.
- * - The application calls {@link #poll(Duration)} without committing, which also implicitly acknowledges all of
- * the delivered records and commits the acknowledgements to Kafka asynchronously. In this case, no exception is
- * thrown by a failure to commit the acknowledgements.
- * - The application calls {@link #close()} which releases any acquired records without acknowledgement.
- *
- *
- * The consumer guarantees that the records returned in the {@code ConsumerRecords} object for a specific topic-partition
- * are in order of increasing offset. For each topic-partition, Kafka guarantees that acknowledgements for the records
- * in a batch are performed atomically. This makes error handling significantly more straightforward because there can be
- * one error code per partition.
- *
- *
Usage Examples
- * The share consumer APIs offer flexibility to cover a variety of consumption use cases. Here are some examples to
- * demonstrate how to use them.
- *
- * Acknowledging a batch of records (implicit acknowledgement)
- * This example demonstrates implicit acknowledgement using {@link #poll(Duration)} to acknowledge the records which
- * were delivered in the previous poll. All the records delivered are implicitly marked as successfully consumed and
- * acknowledged synchronously with Kafka as the consumer fetches more records.
- *
- * Properties props = new Properties();
- * props.setProperty("bootstrap.servers", "localhost:9092");
- * props.setProperty("group.id", "test");
- * props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);
- * consumer.subscribe(Arrays.asList("foo"));
- * while (true) {
- * ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
- * for (ConsumerRecord<String, String> record : records) {
- * System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
- * doProcessing(record);
- * }
- * }
- *
- *
- * Alternatively, you can use {@link #commitSync()} or {@link #commitAsync()} to commit the acknowledgements, but this is
- * slightly less efficient because there is an additional request sent to Kafka.
- *
- * Properties props = new Properties();
- * props.setProperty("bootstrap.servers", "localhost:9092");
- * props.setProperty("group.id", "test");
- * props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);
- * consumer.subscribe(Arrays.asList("foo"));
- * while (true) {
- * ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
- * for (ConsumerRecord<String, String> record : records) {
- * System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
- * doProcessing(record);
- * }
- * consumer.commitSync();
- * }
- *
- *
- * Per-record acknowledgement (explicit acknowledgement)
- * This example demonstrates using different acknowledgement types depending on the outcome of processing the records.
- *
- * Properties props = new Properties();
- * props.setProperty("bootstrap.servers", "localhost:9092");
- * props.setProperty("group.id", "test");
- * props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);
- * consumer.subscribe(Arrays.asList("foo"));
- * while (true) {
- * ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
- * for (ConsumerRecord<String, String> record : records) {
- * try {
- * doProcessing(record);
- * consumer.acknowledge(record, AcknowledgeType.ACCEPT);
- * } catch (Exception e) {
- * consumer.acknowledge(record, AcknowledgeType.REJECT);
- * }
- * }
- * consumer.commitSync();
- * }
- *
- *
- * Each record processed is separately acknowledged using a call to {@link #acknowledge(ConsumerRecord, AcknowledgeType)}.
- * The {@link AcknowledgeType} argument indicates whether the record was processed successfully or not. In this case,
- * the bad records are rejected meaning that they’re not eligible for further delivery attempts. For a permanent error
- * such as a semantic error, this is appropriate. For a transient error which might not affect a subsequent processing
- * attempt, {@link AcknowledgeType#RELEASE} is more appropriate because the record remains eligible for further delivery attempts.
- *
- * The calls to {@link #acknowledge(ConsumerRecord, AcknowledgeType)} are simply updating local information in the consumer.
- * It is only once {@link #commitSync()} is called that the acknowledgements are committed by sending the new state
- * information to Kafka.
- *
- *
Per-record acknowledgement, ending processing of the batch on an error (explicit acknowledgement)
- * This example demonstrates ending processing of a batch of records on the first error.
- *
- * Properties props = new Properties();
- * props.setProperty("bootstrap.servers", "localhost:9092");
- * props.setProperty("group.id", "test");
- * props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- * KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props);
- * consumer.subscribe(Arrays.asList("foo"));
- * while (true) {
- * ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
- * for (ConsumerRecord<String, String> record : records) {
- * try {
- * doProcessing(record);
- * consumer.acknowledge(record, AcknowledgeType.ACCEPT);
- * } catch (Exception e) {
- * consumer.acknowledge(record, AcknowledgeType.REJECT);
- * break;
- * }
- * }
- * consumer.commitSync();
- * }
- *
- * There are the following cases in this example:
- *
- * - The batch contains no records, in which case the application just polls again. The call to {@link #commitSync()}
- * just does nothing because the batch was empty.
- * - All of the records in the batch are processed successfully. The calls to {@link #acknowledge(ConsumerRecord, AcknowledgeType)}
- * specifying {@code AcknowledgeType.ACCEPT} mark all records in the batch as successfully processed.
- * - One of the records encounters an exception. The call to {@link #acknowledge(ConsumerRecord, AcknowledgeType)} specifying
- * {@code AcknowledgeType.REJECT} rejects that record. Earlier records in the batch have already been marked as successfully
- * processed. The call to {@link #commitSync()} commits the acknowledgements, but the records after the failed record
- * remain acquired as part of the same delivery attempt and will be presented to the application in response to another poll.
- *
- *
- * Reading Transactional Records
- * The way that share groups handle transactional records is controlled by the {@code group.share.isolation.level}
- * configuration property. In a share group, the isolation level applies to the entire share group, not just individual
- * consumers.
- *
- * In read_uncommitted isolation level, the share group consumes all non-transactional and transactional
- * records. The consumption is bounded by the high-water mark.
- *
- * In read_committed isolation level (not yet supported), the share group only consumes non-transactional
- * records and committed transactional records. The set of records which are eligible to become in-flight records are
- * non-transactional records and committed transactional records only. The consumption is bounded by the last stable
- * offset, so an open transaction blocks the progress of the share group with read_committed isolation level.
- *
- *
- * The consumer is NOT thread-safe. It is the responsibility of the user to ensure that multithreaded access
- * is properly synchronized. Unsynchronized access will result in {@link java.util.ConcurrentModificationException}.
- *
- * The only exception to this rule is {@link #wakeup()} which can safely be used from an external thread to
- * interrupt an active operation. In this case, a {@link org.apache.kafka.common.errors.WakeupException} will be
- * thrown from the thread blocking on the operation. This can be used to shut down the consumer from another thread.
- * The following snippet shows the typical pattern:
- *
- *
- * public class KafkaShareConsumerRunner implements Runnable {
- * private final AtomicBoolean closed = new AtomicBoolean(false);
- * private final KafkaShareConsumer consumer;
- *
- * public KafkaShareConsumerRunner(KafkaShareConsumer consumer) {
- * this.consumer = consumer;
- * }
- *
- * {@literal}@Override
- * public void run() {
- * try {
- * consumer.subscribe(Arrays.asList("topic"));
- * while (!closed.get()) {
- * ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));
- * // Handle new records
- * }
- * } catch (WakeupException e) {
- * // Ignore exception if closing
- * if (!closed.get()) throw e;
- * } finally {
- * consumer.close();
- * }
- * }
- *
- * // Shutdown hook which can be called from a separate thread
- * public void shutdown() {
- * closed.set(true);
- * consumer.wakeup();
- * }
- * }
- *
- *
- * Then in a separate thread, the consumer can be shutdown by setting the closed flag and waking up the consumer.
- *
- * closed.set(true);
- * consumer.wakeup();
- *
- *
- *
- * Note that while it is possible to use thread interrupts instead of {@link #wakeup()} to abort a blocking operation
- * (in which case, {@link InterruptException} will be raised), we discourage their use since they may cause a clean
- * shutdown of the consumer to be aborted. Interrupts are mainly supported for those cases where using {@link #wakeup()}
- * is impossible, such as when a consumer thread is managed by code that is unaware of the Kafka client.
- *
- * We have intentionally avoided implementing a particular threading model for processing. Various options for
- * multithreaded processing are possible, of which the most straightforward is to dedicate a thread to each consumer.
- */
-@InterfaceStability.Evolving
-public class KafkaShareConsumer implements ShareConsumer {
-
- private final static ShareConsumerDelegateCreator CREATOR = new ShareConsumerDelegateCreator();
-
- private final ShareConsumerDelegate delegate;
-
- /**
- * A consumer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings
- * are documented here. Values can be
- * either strings or objects of the appropriate type (for example a numeric configuration would accept either the
- * string "42" or the integer 42).
- *
- * Valid configuration strings are documented at {@link ConsumerConfig}.
- *
- * Note: after creating a {@code KafkaShareConsumer} you must always {@link #close()} it to avoid resource leaks.
- *
- * @param configs The consumer configs
- */
- public KafkaShareConsumer(Map configs) {
- this(configs, null, null);
- }
-
- /**
- * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration.
- *
- * Valid configuration strings are documented at {@link ConsumerConfig}.
- *
- * Note: after creating a {@code KafkaShareConsumer} you must always {@link #close()} it to avoid resource leaks.
- *
- * @param properties The consumer configuration properties
- */
- public KafkaShareConsumer(Properties properties) {
- this(properties, null, null);
- }
-
- /**
- * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration, and a
- * key and a value {@link Deserializer}.
- *
- * Valid configuration strings are documented at {@link ConsumerConfig}.
- *
- * Note: after creating a {@code KafkaShareConsumer} you must always {@link #close()} it to avoid resource leaks.
- *
- * @param properties The consumer configuration properties
- * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
- * won't be called in the consumer when the deserializer is passed in directly.
- * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
- * won't be called in the consumer when the deserializer is passed in directly.
- */
- public KafkaShareConsumer(Properties properties,
- Deserializer keyDeserializer,
- Deserializer valueDeserializer) {
- this(propsToMap(properties), keyDeserializer, valueDeserializer);
- }
-
- /**
- * A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value {@link Deserializer}.
- *
- * Valid configuration strings are documented at {@link ConsumerConfig}.
- *
- * Note: after creating a {@code KafkaShareConsumer} you must always {@link #close()} it to avoid resource leaks.
- *
- * @param configs The consumer configs
- * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
- * won't be called in the consumer when the deserializer is passed in directly.
- * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
- * won't be called in the consumer when the deserializer is passed in directly.
- */
- public KafkaShareConsumer(Map configs,
- Deserializer keyDeserializer,
- Deserializer valueDeserializer) {
- this(new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(configs, keyDeserializer, valueDeserializer)),
- keyDeserializer, valueDeserializer);
- }
-
- public KafkaShareConsumer(ConsumerConfig config,
- Deserializer keyDeserializer,
- Deserializer valueDeserializer) {
- delegate = CREATOR.create(config, keyDeserializer, valueDeserializer);
- }
-
- KafkaShareConsumer(final LogContext logContext,
- final String clientId,
- final String groupId,
- final ConsumerConfig config,
- final Deserializer keyDeserializer,
- final Deserializer valueDeserializer,
- final Time time,
- final KafkaClient client,
- final SubscriptionState subscriptions,
- final ConsumerMetadata metadata) {
- delegate = CREATOR.create(
- logContext, clientId, groupId, config, keyDeserializer, valueDeserializer,
- time, client, subscriptions, metadata);
- }
-
- /**
- * Get the current subscription. Will return the same topics used in the most recent call to
- * {@link #subscribe(Collection)}, or an empty set if no such call has been made.
- *
- * @return The set of topics currently subscribed to
- */
- @Override
- public Set subscription() {
- return delegate.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. If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
- *
- *
- * As part of group management, the coordinator will keep track of the list of consumers that belong to a particular
- * group and will trigger a rebalance operation if any one of the following events are triggered:
- *
- * - A member joins or leaves the share group
- *
- An existing member of the share group is shut down or fails
- *
- The number of partitions changes for any of the subscribed topics
- *
- A subscribed topic is created or deleted
- *
- *
- * @param topics The list of topics to subscribe to
- *
- * @throws IllegalArgumentException if topics is null or contains null or empty elements
- * @throws KafkaException for any other unrecoverable errors
- */
- @Override
- public void subscribe(Collection topics) {
- delegate.subscribe(topics);
- }
-
- /**
- * Unsubscribe from topics currently subscribed with {@link #subscribe(Collection)}.
- *
- * @throws KafkaException for any other unrecoverable errors
- */
- @Override
- public void unsubscribe() {
- delegate.unsubscribe();
- }
-
- /**
- * Fetch data for the topics specified using {@link #subscribe(Collection)}. It is an error to not have
- * subscribed to any topics before polling for data.
- *
- *
- * This method returns immediately if there are records available. Otherwise, it will await the passed timeout.
- * If the timeout expires, an empty record set will be returned.
- *
- * @param timeout The maximum time to block (must not be greater than {@link Long#MAX_VALUE} milliseconds)
- *
- * @return map of topic to records since the last fetch for the subscribed list of topics
- *
- * @throws AuthenticationException if authentication fails. See the exception for more details
- * @throws AuthorizationException if caller lacks Read access to any of the subscribed
- * topics or to the share group. See the exception for more details
- * @throws IllegalArgumentException if the timeout value is negative
- * @throws IllegalStateException if the consumer is not subscribed to any topics
- * @throws ArithmeticException if the timeout is greater than {@link Long#MAX_VALUE} milliseconds.
- * @throws InvalidTopicException if the current subscription contains any invalid
- * topic (per {@link org.apache.kafka.common.internals.Topic#validate(String)})
- * @throws WakeupException if {@link #wakeup()} is called before or while this method is called
- * @throws InterruptException if the calling thread is interrupted before or while this method is called
- * @throws KafkaException for any other unrecoverable errors
- */
- @Override
- public ConsumerRecords poll(Duration timeout) {
- return delegate.poll(timeout);
- }
-
- /**
- * Acknowledge successful delivery of a record returned on the last {@link #poll(Duration)} call.
- * The acknowledgement is committed on the next {@link #commitSync()}, {@link #commitAsync()} or
- * {@link #poll(Duration)} call.
- *
- * @param record The record to acknowledge
- *
- * @throws IllegalStateException if the record is not waiting to be acknowledged, or the consumer has already
- * used implicit acknowledgement
- */
- @Override
- public void acknowledge(ConsumerRecord record) {
- delegate.acknowledge(record);
- }
-
- /**
- * Acknowledge delivery of a record returned on the last {@link #poll(Duration)} call indicating whether
- * it was processed successfully. The acknowledgement is committed on the next {@link #commitSync()},
- * {@link #commitAsync()} or {@link #poll(Duration)} call. By using this method, the consumer is using
- * explicit acknowledgement.
- *
- * @param record The record to acknowledge
- * @param type The acknowledgement type which indicates whether it was processed successfully
- *
- * @throws IllegalStateException if the record is not waiting to be acknowledged, or the consumer has already
- * used implicit acknowledgement
- */
- @Override
- public void acknowledge(ConsumerRecord record, AcknowledgeType type) {
- delegate.acknowledge(record, type);
- }
-
- /**
- * Commit the acknowledgements for the records returned. If the consumer is using explicit acknowledgement,
- * the acknowledgements to commit have been indicated using {@link #acknowledge(ConsumerRecord)} or
- * {@link #acknowledge(ConsumerRecord, AcknowledgeType)}. If the consumer is using implicit acknowledgement,
- * all the records returned by the latest call to {@link #poll(Duration)} are acknowledged.
- *
- *
- * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is
- * encountered (in which case it is thrown to the caller), or the timeout specified by {@code default.api.timeout.ms}
- * expires.
- *
- * @return A map of the results for each topic-partition for which delivery was acknowledged.
- * If the acknowledgement failed for a topic-partition, an exception is present.
- *
- * @throws WakeupException if {@link #wakeup()} is called before or while this method is called
- * @throws InterruptException if the thread is interrupted while blocked
- * @throws KafkaException for any other unrecoverable errors
- */
- @Override
- public Map> commitSync() {
- return delegate.commitSync();
- }
-
- /**
- * Commit the acknowledgements for the records returned. If the consumer is using explicit acknowledgement,
- * the acknowledgements to commit have been indicated using {@link #acknowledge(ConsumerRecord)} or
- * {@link #acknowledge(ConsumerRecord, AcknowledgeType)}. If the consumer is using implicit acknowledgement,
- * all the records returned by the latest call to {@link #poll(Duration)} are acknowledged.
-
- *
- * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is
- * encountered (in which case it is thrown to the caller), or the timeout expires.
- *
- * @param timeout The maximum amount of time to await completion of the acknowledgement
- *
- * @return A map of the results for each topic-partition for which delivery was acknowledged.
- * If the acknowledgement failed for a topic-partition, an exception is present.
- *
- * @throws IllegalArgumentException if the {@code timeout} is negative
- * @throws WakeupException if {@link #wakeup()} is called before or while this method is called
- * @throws InterruptException if the thread is interrupted while blocked
- * @throws KafkaException for any other unrecoverable errors
- */
- @Override
- public Map> commitSync(Duration timeout) {
- return delegate.commitSync(timeout);
- }
-
- /**
- * Commit the acknowledgements for the records returned. If the consumer is using explicit acknowledgement,
- * the acknowledgements to commit have been indicated using {@link #acknowledge(ConsumerRecord)} or
- * {@link #acknowledge(ConsumerRecord, AcknowledgeType)}. If the consumer is using implicit acknowledgement,
- * all the records returned by the latest call to {@link #poll(Duration)} are acknowledged.
- *
- * @throws KafkaException for any other unrecoverable errors
- */
- @Override
- public void commitAsync() {
- delegate.commitAsync();
- }
-
- /**
- * Sets the acknowledgement commit callback which can be used to handle acknowledgement completion.
- *
- * @param callback The acknowledgement commit callback
- */
- @Override
- public void setAcknowledgementCommitCallback(AcknowledgementCommitCallback callback) {
- delegate.setAcknowledgementCommitCallback(callback);
- }
-
- /**
- * Determines the client's unique client instance ID used for telemetry. This ID is unique to
- * this specific client instance and will not change after it is initially generated.
- * The ID is useful for correlating client operations with telemetry sent to the broker and
- * to its eventual monitoring destinations.
- *
- * If telemetry is enabled, this will first require a connection to the cluster to generate
- * the unique client instance ID. This method waits up to {@code timeout} for the consumer
- * client to complete the request.
- *
- * Client telemetry is controlled by the {@link ConsumerConfig#ENABLE_METRICS_PUSH_CONFIG}
- * configuration option.
- *
- * @param timeout The maximum time to wait for consumer client to determine its client instance ID.
- * The value must be non-negative. Specifying a timeout of zero means do not
- * wait for the initial request to complete if it hasn't already.
- *
- * @return The client's assigned instance id used for metrics collection.
- *
- * @throws IllegalArgumentException if the {@code timeout} is negative
- * @throws IllegalStateException if telemetry is not enabled
- * @throws WakeupException if {@link #wakeup()} is called before or while this method is called
- * @throws InterruptException if the thread is interrupted while blocked
- * @throws KafkaException if an unexpected error occurs while trying to determine the client
- * instance ID, though this error does not necessarily imply the
- * consumer client is otherwise unusable
- */
- @Override
- public Uuid clientInstanceId(Duration timeout) {
- return delegate.clientInstanceId(timeout);
- }
-
- /**
- * Get the metrics kept by the consumer
- */
- @Override
- public Map metrics() {
- return delegate.metrics();
- }
-
- /**
- * Close the consumer, waiting for up to the default timeout of 30 seconds for any needed cleanup.
- * This will commit acknowledgements if possible within the default timeout.
- * See {@link #close(Duration)} for details. Note that {@link #wakeup()} cannot be used to interrupt close.
- *
- * @throws WakeupException if {@link #wakeup()} is called before or while this method is called
- * @throws InterruptException if the thread is interrupted before or while this method is called
- * @throws KafkaException for any other error during close
- */
- @Override
- public void close() {
- delegate.close();
- }
-
- /**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete acknowledgements and leave the group.
- * If the consumer is unable to complete acknowledgements and gracefully leave the group
- * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
- * used to interrupt close.
- *
- * @param timeout The maximum time to wait for consumer to close gracefully. The value must be
- * non-negative. Specifying a timeout of zero means do not wait for pending requests to complete.
- *
- * @throws IllegalArgumentException if the {@code timeout} is negative
- * @throws WakeupException if {@link #wakeup()} is called before or while this method is called
- * @throws InterruptException if the thread is interrupted before or while this method is called
- * @throws KafkaException for any other error during close
- */
- @Override
- public void close(Duration timeout) {
- delegate.close(timeout);
- }
-
- /**
- * Wake up the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link WakeupException}.
- * If no thread is blocking in a method which can throw {@link WakeupException},
- * the next call to such a method will raise it instead.
- */
- @Override
- public void wakeup() {
- delegate.wakeup();
- }
-
- // Functions below are for testing only
- String clientId() {
- return delegate.clientId();
- }
-
- Metrics metricsRegistry() {
- return delegate.metricsRegistry();
- }
-}
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 600f8bbd07ef4..27faa80c65421 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
@@ -253,7 +253,7 @@ public synchronized ConsumerRecords poll(final Duration timeout) {
}
}
- toClear.forEach(records::remove);
+ toClear.forEach(p -> this.records.remove(p));
return new ConsumerRecords<>(results);
}
@@ -263,7 +263,7 @@ public synchronized void addRecord(ConsumerRecord record) {
Set currentAssigned = this.subscriptions.assignedPartitions();
if (!currentAssigned.contains(tp))
throw new IllegalStateException("Cannot add records for a partition that is not assigned to the consumer");
- List> recs = records.computeIfAbsent(tp, k -> new ArrayList<>());
+ List> recs = this.records.computeIfAbsent(tp, k -> new ArrayList<>());
recs.add(record);
}
@@ -286,7 +286,8 @@ public synchronized void setOffsetsException(KafkaException exception) {
@Override
public synchronized void commitAsync(Map offsets, OffsetCommitCallback callback) {
ensureNotClosed();
- committed.putAll(offsets);
+ for (Map.Entry entry : offsets.entrySet())
+ committed.put(entry.getKey(), entry.getValue());
if (callback != null) {
callback.onComplete(offsets, null);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java
deleted file mode 100644
index 212e2ae390d5f..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockShareConsumer.java
+++ /dev/null
@@ -1,170 +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.clients.consumer;
-
-import org.apache.kafka.clients.consumer.internals.SubscriptionState;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.Metric;
-import org.apache.kafka.common.MetricName;
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.LogContext;
-
-import java.time.Duration;
-import java.util.ArrayList;
-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.Optional;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.DEFAULT_CLOSE_TIMEOUT_MS;
-
-/**
- * A mock of the {@link ShareConsumer} interface you can use for testing code that uses Kafka. This class is not
- * thread-safe .
- */
-public class MockShareConsumer implements ShareConsumer {
-
- private final SubscriptionState subscriptions;
- private final AtomicBoolean wakeup;
-
- private final Map>> records;
-
- private boolean closed;
- private Uuid clientInstanceId;
-
- public MockShareConsumer() {
- this.subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
- this.records = new HashMap<>();
- this.closed = false;
- this.wakeup = new AtomicBoolean(false);
- }
-
- @Override
- public synchronized Set subscription() {
- ensureNotClosed();
- return subscriptions.subscription();
- }
-
- @Override
- public synchronized void subscribe(Collection topics) {
- ensureNotClosed();
- subscriptions.subscribe(new HashSet<>(topics), Optional.empty());
- }
-
- @Override
- public synchronized void unsubscribe() {
- ensureNotClosed();
- subscriptions.unsubscribe();
- }
-
- @Override
- public synchronized ConsumerRecords poll(Duration timeout) {
- ensureNotClosed();
-
- final Map>> results = new HashMap<>();
- for (Map.Entry>> entry : records.entrySet()) {
- final List> recs = entry.getValue();
- for (final ConsumerRecord rec : recs) {
- results.computeIfAbsent(entry.getKey(), partition -> new ArrayList<>()).add(rec);
- }
- }
-
- records.clear();
- return new ConsumerRecords<>(results);
- }
-
- @Override
- public synchronized void acknowledge(ConsumerRecord record) {
- }
-
- @Override
- public synchronized void acknowledge(ConsumerRecord record, AcknowledgeType type) {
- }
-
- @Override
- public synchronized Map> commitSync() {
- return new HashMap<>();
- }
-
- @Override
- public synchronized Map> commitSync(Duration timeout) {
- return new HashMap<>();
- }
-
- @Override
- public synchronized void commitAsync() {
- }
-
- @Override
- public void setAcknowledgementCommitCallback(AcknowledgementCommitCallback callback) {
- }
-
- public synchronized void setClientInstanceId(final Uuid clientInstanceId) {
- this.clientInstanceId = clientInstanceId;
- }
-
- @Override
- public synchronized Uuid clientInstanceId(Duration timeout) {
- if (clientInstanceId == null) {
- throw new UnsupportedOperationException("clientInstanceId not set");
- }
-
- return clientInstanceId;
- }
-
- @Override
- public synchronized Map metrics() {
- ensureNotClosed();
- return Collections.emptyMap();
- }
-
- @Override
- public synchronized void close() {
- close(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));
- }
-
- @Override
- public synchronized void close(Duration timeout) {
- closed = true;
- }
-
- @Override
- public synchronized void wakeup() {
- wakeup.set(true);
- }
-
- public synchronized void addRecord(ConsumerRecord record) {
- ensureNotClosed();
- TopicPartition tp = new TopicPartition(record.topic(), record.partition());
- if (!subscriptions.subscription().contains(record.topic()))
- throw new IllegalStateException("Cannot add records for a topics that is not subscribed by the consumer");
- List> recs = records.computeIfAbsent(tp, k -> new ArrayList<>());
- recs.add(record);
- }
-
- private void ensureNotClosed() {
- if (closed)
- throw new IllegalStateException("This consumer has already been closed.");
- }
-}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java
index 6bc064006fe6f..e95d4f1efd7b0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java
@@ -170,7 +170,7 @@ private void assignRanges(TopicAssignmentState assignmentState,
private void assignWithRackMatching(Collection assignmentStates,
Map> assignment) {
- assignmentStates.stream().collect(Collectors.groupingBy(t -> t.consumers)).forEach((consumers, states) ->
+ assignmentStates.stream().collect(Collectors.groupingBy(t -> t.consumers)).forEach((consumers, states) -> {
states.stream().collect(Collectors.groupingBy(t -> t.partitionRacks.size())).forEach((numPartitions, coPartitionedStates) -> {
if (coPartitionedStates.size() > 1)
assignCoPartitionedWithRackMatching(consumers, numPartitions, coPartitionedStates, assignment);
@@ -179,8 +179,8 @@ private void assignWithRackMatching(Collection assignmentS
if (state.needsRackAwareAssignment)
assignRanges(state, state::racksMatch, assignment);
}
- })
- );
+ });
+ });
}
private void assignCoPartitionedWithRackMatching(LinkedHashMap> consumers,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java
deleted file mode 100644
index 8ac4198c70df3..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java
+++ /dev/null
@@ -1,115 +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.clients.consumer;
-
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.Metric;
-import org.apache.kafka.common.MetricName;
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.annotation.InterfaceStability;
-
-import java.io.Closeable;
-import java.time.Duration;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-
-/**
- * @see KafkaShareConsumer
- * @see MockShareConsumer
- */
-@InterfaceStability.Evolving
-public interface ShareConsumer extends Closeable {
-
- /**
- * @see KafkaShareConsumer#subscription()
- */
- Set subscription();
-
- /**
- * @see KafkaShareConsumer#subscribe(Collection)
- */
- void subscribe(Collection topics);
-
- /**
- * @see KafkaShareConsumer#unsubscribe()
- */
- void unsubscribe();
-
- /**
- * @see KafkaShareConsumer#poll(Duration)
- */
- ConsumerRecords poll(Duration timeout);
-
- /**
- * @see KafkaShareConsumer#acknowledge(ConsumerRecord)
- */
- void acknowledge(ConsumerRecord record);
-
- /**
- * @see KafkaShareConsumer#acknowledge(ConsumerRecord, AcknowledgeType)
- */
- void acknowledge(ConsumerRecord record, AcknowledgeType type);
-
- /**
- * @see KafkaShareConsumer#commitSync()
- */
- Map> commitSync();
-
- /**
- * @see KafkaShareConsumer#commitSync(Duration)
- */
- Map> commitSync(Duration timeout);
-
- /**
- * @see KafkaShareConsumer#commitAsync()
- */
- void commitAsync();
-
- /**
- * @see KafkaShareConsumer#setAcknowledgementCommitCallback(AcknowledgementCommitCallback)
- */
- void setAcknowledgementCommitCallback(AcknowledgementCommitCallback callback);
-
- /**
- * See {@link KafkaShareConsumer#clientInstanceId(Duration)}}
- */
- Uuid clientInstanceId(Duration timeout);
-
- /**
- * @see KafkaShareConsumer#metrics()
- */
- Map metrics();
-
- /**
- * @see KafkaShareConsumer#close()
- */
- void close();
-
- /**
- * @see KafkaShareConsumer#close(Duration)
- */
- void close(Duration timeout);
-
- /**
- * @see KafkaShareConsumer#wakeup()
- */
- void wakeup();
-
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
index 0d3e4a256e2be..7e7350a5946e8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
@@ -236,7 +236,8 @@ static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData) {
topicAssignments.add(topicAssignment);
}
struct.set(TOPIC_PARTITIONS_KEY_NAME, topicAssignments.toArray());
- memberData.generation.ifPresent(integer -> struct.set(GENERATION_KEY_NAME, integer));
+ if (memberData.generation.isPresent())
+ struct.set(GENERATION_KEY_NAME, memberData.generation.get());
ByteBuffer buffer = ByteBuffer.allocate(STICKY_ASSIGNOR_USER_DATA_V1.sizeOf(struct));
STICKY_ASSIGNOR_USER_DATA_V1.write(buffer, struct);
buffer.flip();
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index e6de169f7b33f..6930cd02955d6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -26,6 +26,7 @@
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
+import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.GroupProtocol;
@@ -234,12 +235,12 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
private final SubscriptionState subscriptions;
private final ConsumerMetadata metadata;
- private int metadataVersionSnapshot;
private final Metrics metrics;
private final long retryBackoffMs;
private final int defaultApiTimeoutMs;
private final boolean autoCommitEnabled;
private volatile boolean closed = false;
+ private final List assignors;
private final Optional clientTelemetryReporter;
// to keep from repeatedly scanning subscriptions in poll(), cache the result during metadata updates
@@ -312,7 +313,6 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
this.metadata = metadataFactory.build(config, subscriptions, logContext, clusterResourceListeners);
final List addresses = ClientUtils.parseAndValidateAddresses(config);
metadata.bootstrap(addresses);
- this.metadataVersionSnapshot = metadata.updateVersion();
FetchMetricsManager fetchMetricsManager = createFetchMetricsManager(metrics);
FetchConfig fetchConfig = new FetchConfig(config);
@@ -331,8 +331,7 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
apiVersions,
metrics,
fetchMetricsManager,
- clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null),
- backgroundEventHandler);
+ clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null));
this.offsetCommitCallbackInvoker = new OffsetCommitCallbackInvoker(interceptors);
this.asyncCommitFenced = new AtomicBoolean(false);
this.groupMetadata.set(initializeGroupMetadata(config, groupRebalanceConfig));
@@ -374,6 +373,10 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
rebalanceListenerInvoker
);
this.backgroundEventReaper = backgroundEventReaperFactory.build(logContext);
+ this.assignors = ConsumerPartitionAssignor.getAssignorInstances(
+ config.getList(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG),
+ config.originals(Collections.singletonMap(ConsumerConfig.CLIENT_ID_CONFIG, clientId))
+ );
// The FetchCollector is only used on the application thread.
this.fetchCollector = fetchCollectorFactory.build(logContext,
@@ -421,6 +424,7 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
ConsumerMetadata metadata,
long retryBackoffMs,
int defaultApiTimeoutMs,
+ List assignors,
String groupId,
boolean autoCommitEnabled) {
this.log = logContext.logger(getClass());
@@ -437,11 +441,11 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
this.metrics = metrics;
this.groupMetadata.set(initializeGroupMetadata(groupId, Optional.empty()));
this.metadata = metadata;
- this.metadataVersionSnapshot = metadata.updateVersion();
this.retryBackoffMs = retryBackoffMs;
this.defaultApiTimeoutMs = defaultApiTimeoutMs;
this.deserializers = deserializers;
this.applicationEventHandler = applicationEventHandler;
+ this.assignors = assignors;
this.kafkaConsumerMetrics = new KafkaConsumerMetrics(metrics, "consumer");
this.clientTelemetryReporter = Optional.empty();
this.autoCommitEnabled = autoCommitEnabled;
@@ -456,7 +460,8 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
Deserializer valueDeserializer,
KafkaClient client,
SubscriptionState subscriptions,
- ConsumerMetadata metadata) {
+ ConsumerMetadata metadata,
+ List assignors) {
this.log = logContext.logger(getClass());
this.subscriptions = subscriptions;
this.clientId = config.getString(ConsumerConfig.CLIENT_ID_CONFIG);
@@ -467,10 +472,10 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
this.time = time;
this.metrics = new Metrics(time);
this.metadata = metadata;
- this.metadataVersionSnapshot = metadata.updateVersion();
this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
this.defaultApiTimeoutMs = config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
this.deserializers = new Deserializers<>(keyDeserializer, valueDeserializer);
+ this.assignors = assignors;
this.clientTelemetryReporter = Optional.empty();
ConsumerMetrics metricsRegistry = new ConsumerMetrics(CONSUMER_METRIC_GROUP_PREFIX);
@@ -505,9 +510,7 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
time,
config,
logContext,
- client,
- metadata,
- backgroundEventHandler
+ client
);
this.offsetCommitCallbackInvoker = new OffsetCommitCallbackInvoker(interceptors);
this.asyncCommitFenced = new AtomicBoolean(false);
@@ -1475,11 +1478,12 @@ public void assign(Collection partitions) {
}
/**
- *
+ * TODO: remove this when we implement the KIP-848 protocol.
*
- * This function evaluates the regex that the consumer subscribed to
- * against the list of topic names from metadata, and updates
- * the list of topics in subscription state accordingly
+ *
+ * The contents of this method are shamelessly stolen from
+ * {@link ConsumerCoordinator#updatePatternSubscription(Cluster)} and are used here because we won't have access
+ * to a {@link ConsumerCoordinator} in this code. Perhaps it could be moved to a ConsumerUtils class?
*
* @param cluster Cluster from which we get the topics
*/
@@ -1489,7 +1493,7 @@ private void updatePatternSubscription(Cluster cluster) {
.collect(Collectors.toSet());
if (subscriptions.subscribeFromPattern(topicsToSubscribe)) {
applicationEventHandler.add(new SubscriptionChangeEvent());
- this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+ metadata.requestUpdateForNewTopics();
}
}
@@ -1683,6 +1687,12 @@ private boolean initWithCommittedOffsetsIfNeeded(Timer timer) {
}
}
+ private void throwIfNoAssignorsConfigured() {
+ if (assignors.isEmpty())
+ throw new IllegalStateException("Must configure at least one partition assigner class name to " +
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG + " configuration property");
+ }
+
private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAndMetadata offsetAndMetadata) {
if (offsetAndMetadata != null)
offsetAndMetadata.leaderEpoch().ifPresent(epoch -> metadata.updateLastSeenEpochIfNewer(topicPartition, epoch));
@@ -1770,6 +1780,7 @@ private void subscribeInternal(Pattern pattern, Optional topics, Optional currentTopicPartitions = new HashSet<>();
@@ -1805,7 +1818,7 @@ private void subscribeInternal(Collection topics, Optional(topics), listener))
- this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+ metadata.requestUpdateForNewTopics();
// Trigger subscribe event to effectively join the group if not already part of it,
// or just send the new subscription to the broker.
@@ -1984,12 +1997,9 @@ SubscriptionState subscriptions() {
}
private void maybeUpdateSubscriptionMetadata() {
- if (this.metadataVersionSnapshot < metadata.updateVersion()) {
- this.metadataVersionSnapshot = metadata.updateVersion();
- if (subscriptions.hasPatternSubscription()) {
- updatePatternSubscription(metadata.fetch());
- }
+ if (subscriptions.hasPatternSubscription()) {
+ updatePatternSubscription(metadata.fetch());
}
}
-}
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
index 000797dba0918..f6a11fe4b7d58 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
@@ -69,7 +69,6 @@
import static org.apache.kafka.common.protocol.Errors.COORDINATOR_LOAD_IN_PROGRESS;
public class CommitRequestManager implements RequestManager, MemberStateListener {
- private final Time time;
private final SubscriptionState subscriptions;
private final LogContext logContext;
private final Logger log;
@@ -134,7 +133,6 @@ public CommitRequestManager(
final OptionalDouble jitter,
final Metrics metrics) {
Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets");
- this.time = time;
this.logContext = logContext;
this.log = logContext.logger(getClass());
this.pendingRequests = new PendingRequests();
@@ -207,13 +205,6 @@ private static long findMinTime(final Collection request
.orElse(Long.MAX_VALUE);
}
- private KafkaException maybeWrapAsTimeoutException(Throwable t) {
- if (t instanceof TimeoutException)
- return (TimeoutException) t;
- else
- return new TimeoutException(t);
- }
-
/**
* Generate a request to commit consumed offsets. Add the request to the queue of pending
* requests to be sent out on the next call to {@link #poll(long)}. If there are empty
@@ -254,7 +245,7 @@ public void maybeAutoCommitAsync() {
if (autoCommitEnabled() && autoCommitState.get().shouldAutoCommit()) {
OffsetCommitRequestState requestState = createOffsetCommitRequest(
subscriptions.allConsumed(),
- Long.MAX_VALUE);
+ Optional.empty());
CompletableFuture> result = requestAutoCommit(requestState);
// Reset timer to the interval (even if no request was generated), but ensure that if
// the request completes with a retriable error, the timer is reset to send the next
@@ -303,14 +294,14 @@ private void maybeResetTimerWithBackoff(final CompletableFuture maybeAutoCommitSyncBeforeRevocation(final long deadlineMs) {
+ public CompletableFuture maybeAutoCommitSyncBeforeRevocation(final long retryExpirationTimeMs) {
if (!autoCommitEnabled()) {
return CompletableFuture.completedFuture(null);
}
CompletableFuture result = new CompletableFuture<>();
OffsetCommitRequestState requestState =
- createOffsetCommitRequest(subscriptions.allConsumed(), deadlineMs);
+ createOffsetCommitRequest(subscriptions.allConsumed(), Optional.of(retryExpirationTimeMs));
autoCommitSyncBeforeRevocationWithRetries(requestState, result);
return result;
}
@@ -323,9 +314,9 @@ private void autoCommitSyncBeforeRevocationWithRetries(OffsetCommitRequestState
result.complete(null);
} else {
if (error instanceof RetriableException || isStaleEpochErrorAndValidEpochAvailable(error)) {
- if (requestAttempt.isExpired()) {
+ if (error instanceof TimeoutException && requestAttempt.isExpired) {
log.debug("Auto-commit sync before revocation timed out and won't be retried anymore");
- result.completeExceptionally(maybeWrapAsTimeoutException(error));
+ result.completeExceptionally(error);
} else if (error instanceof UnknownTopicOrPartitionException) {
log.debug("Auto-commit sync before revocation failed because topic or partition were deleted");
result.completeExceptionally(error);
@@ -376,7 +367,7 @@ public CompletableFuture commitAsync(final Map asyncCommitResult = new CompletableFuture<>();
@@ -394,26 +385,28 @@ public CompletableFuture commitAsync(final Map commitSync(final Map offsets,
- final long deadlineMs) {
+ final long retryExpirationTimeMs) {
CompletableFuture result = new CompletableFuture<>();
- OffsetCommitRequestState requestState = createOffsetCommitRequest(offsets, deadlineMs);
+ OffsetCommitRequestState requestState = createOffsetCommitRequest(
+ offsets,
+ Optional.of(retryExpirationTimeMs));
commitSyncWithRetries(requestState, result);
return result;
}
private OffsetCommitRequestState createOffsetCommitRequest(final Map offsets,
- final long deadlineMs) {
+ final Optional expirationTimeMs) {
return jitter.isPresent() ?
new OffsetCommitRequestState(
offsets,
groupId,
groupInstanceId,
- deadlineMs,
+ expirationTimeMs,
retryBackoffMs,
retryBackoffMaxMs,
jitter.getAsDouble(),
@@ -422,7 +415,7 @@ private OffsetCommitRequestState createOffsetCommitRequest(final Map> fetchOffsets(
final Set partitions,
- final long deadlineMs) {
+ final long expirationTimeMs) {
if (partitions.isEmpty()) {
return CompletableFuture.completedFuture(Collections.emptyMap());
}
CompletableFuture> result = new CompletableFuture<>();
- OffsetFetchRequestState request = createOffsetFetchRequest(partitions, deadlineMs);
+ OffsetFetchRequestState request = createOffsetFetchRequest(partitions, expirationTimeMs);
fetchOffsetsWithRetries(request, result);
return result;
}
private OffsetFetchRequestState createOffsetFetchRequest(final Set partitions,
- final long deadlineMs) {
+ final long expirationTimeMs) {
return jitter.isPresent() ?
new OffsetFetchRequestState(
partitions,
retryBackoffMs,
retryBackoffMaxMs,
- deadlineMs,
+ expirationTimeMs,
jitter.getAsDouble(),
memberInfo) :
new OffsetFetchRequestState(
partitions,
retryBackoffMs,
retryBackoffMaxMs,
- deadlineMs,
+ expirationTimeMs,
memberInfo);
}
@@ -523,9 +516,8 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,
result.complete(res);
} else {
if (error instanceof RetriableException || isStaleEpochErrorAndValidEpochAvailable(error)) {
- if (fetchRequest.isExpired()) {
- log.debug("OffsetFetch request for {} timed out and won't be retried anymore", fetchRequest.requestedPartitions);
- result.completeExceptionally(maybeWrapAsTimeoutException(error));
+ if (error instanceof TimeoutException && fetchRequest.isExpired) {
+ result.completeExceptionally(error);
} else {
fetchRequest.resetFuture();
fetchOffsetsWithRetries(fetchRequest, result);
@@ -620,12 +612,12 @@ private class OffsetCommitRequestState extends RetriableRequestState {
OffsetCommitRequestState(final Map offsets,
final String groupId,
final Optional groupInstanceId,
- final long deadlineMs,
+ final Optional expirationTimeMs,
final long retryBackoffMs,
final long retryBackoffMaxMs,
final MemberInfo memberInfo) {
super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs,
- retryBackoffMaxMs, memberInfo, deadlineTimer(time, deadlineMs));
+ retryBackoffMaxMs, memberInfo, expirationTimeMs);
this.offsets = offsets;
this.groupId = groupId;
this.groupInstanceId = groupInstanceId;
@@ -636,13 +628,13 @@ private class OffsetCommitRequestState extends RetriableRequestState {
OffsetCommitRequestState(final Map offsets,
final String groupId,
final Optional groupInstanceId,
- final long deadlineMs,
+ final Optional expirationTimeMs,
final long retryBackoffMs,
final long retryBackoffMaxMs,
final double jitter,
final MemberInfo memberInfo) {
super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2,
- retryBackoffMaxMs, jitter, memberInfo, deadlineTimer(time, deadlineMs));
+ retryBackoffMaxMs, jitter, memberInfo, expirationTimeMs);
this.offsets = offsets;
this.groupId = groupId;
this.groupInstanceId = groupInstanceId;
@@ -788,24 +780,40 @@ void removeRequest() {
* Represents a request that can be retried or aborted, based on member ID and epoch
* information.
*/
- abstract class RetriableRequestState extends TimedRequestState {
+ abstract class RetriableRequestState extends RequestState {
/**
* Member info (ID and epoch) to be included in the request if present.
*/
final MemberInfo memberInfo;
+ /**
+ * Time until which the request should be retried if it fails with retriable
+ * errors. If not present, the request is triggered without waiting for a response or
+ * retrying.
+ */
+ private final Optional expirationTimeMs;
+
+ /**
+ * True if the request expiration time has been reached. This is set when validating the
+ * request expiration on {@link #poll(long)} before sending it. It is used to know if a
+ * request should be retried on TimeoutException.
+ */
+ boolean isExpired;
+
RetriableRequestState(LogContext logContext, String owner, long retryBackoffMs,
- long retryBackoffMaxMs, MemberInfo memberInfo, Timer timer) {
- super(logContext, owner, retryBackoffMs, retryBackoffMaxMs, timer);
+ long retryBackoffMaxMs, MemberInfo memberInfo, Optional expirationTimeMs) {
+ super(logContext, owner, retryBackoffMs, retryBackoffMaxMs);
this.memberInfo = memberInfo;
+ this.expirationTimeMs = expirationTimeMs;
}
// Visible for testing
RetriableRequestState(LogContext logContext, String owner, long retryBackoffMs, int retryBackoffExpBase,
- long retryBackoffMaxMs, double jitter, MemberInfo memberInfo, Timer timer) {
- super(logContext, owner, retryBackoffMs, retryBackoffExpBase, retryBackoffMaxMs, jitter, timer);
+ long retryBackoffMaxMs, double jitter, MemberInfo memberInfo, Optional expirationTimeMs) {
+ super(logContext, owner, retryBackoffMs, retryBackoffExpBase, retryBackoffMaxMs, jitter);
this.memberInfo = memberInfo;
+ this.expirationTimeMs = expirationTimeMs;
}
/**
@@ -820,12 +828,13 @@ abstract class RetriableRequestState extends TimedRequestState {
abstract CompletableFuture future();
/**
- * Complete the request future with a TimeoutException if the request has been sent out
- * at least once and the timeout has been reached.
+ * Complete the request future with a TimeoutException if the request timeout has been
+ * reached, based on the provided current time.
*/
- void maybeExpire() {
- if (numAttempts > 0 && isExpired()) {
+ void maybeExpire(long currentTimeMs) {
+ if (retryTimeoutExpired(currentTimeMs)) {
removeRequest();
+ isExpired = true;
future().completeExceptionally(new TimeoutException(requestDescription() +
" could not complete before timeout expired."));
}
@@ -837,12 +846,11 @@ void maybeExpire() {
NetworkClientDelegate.UnsentRequest buildRequestWithResponseHandling(final AbstractRequest.Builder builder) {
NetworkClientDelegate.UnsentRequest request = new NetworkClientDelegate.UnsentRequest(
builder,
- coordinatorRequestManager.coordinator()
- );
+ coordinatorRequestManager.coordinator());
request.whenComplete(
(response, throwable) -> {
- long completionTimeMs = request.handler().completionTimeMs();
- handleClientResponse(response, throwable, completionTimeMs);
+ long currentTimeMs = request.handler().completionTimeMs();
+ handleClientResponse(response, throwable, currentTimeMs);
});
return request;
}
@@ -867,7 +875,16 @@ private void handleClientResponse(final ClientResponse response,
abstract void onResponse(final ClientResponse response);
+ boolean retryTimeoutExpired(long currentTimeMs) {
+ return expirationTimeMs.isPresent() && expirationTimeMs.get() <= currentTimeMs;
+ }
+
abstract void removeRequest();
+
+ // Visible for testing
+ protected Optional expirationTimeMs() {
+ return expirationTimeMs;
+ }
}
class OffsetFetchRequestState extends RetriableRequestState {
@@ -886,10 +903,10 @@ class OffsetFetchRequestState extends RetriableRequestState {
public OffsetFetchRequestState(final Set partitions,
final long retryBackoffMs,
final long retryBackoffMaxMs,
- final long deadlineMs,
+ final long expirationTimeMs,
final MemberInfo memberInfo) {
super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs,
- retryBackoffMaxMs, memberInfo, deadlineTimer(time, deadlineMs));
+ retryBackoffMaxMs, memberInfo, Optional.of(expirationTimeMs));
this.requestedPartitions = partitions;
this.future = new CompletableFuture<>();
}
@@ -897,11 +914,11 @@ public OffsetFetchRequestState(final Set partitions,
public OffsetFetchRequestState(final Set partitions,
final long retryBackoffMs,
final long retryBackoffMaxMs,
- final long deadlineMs,
+ final long expirationTimeMs,
final double jitter,
final MemberInfo memberInfo) {
super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2,
- retryBackoffMaxMs, jitter, memberInfo, deadlineTimer(time, deadlineMs));
+ retryBackoffMaxMs, jitter, memberInfo, Optional.of(expirationTimeMs));
this.requestedPartitions = partitions;
this.future = new CompletableFuture<>();
}
@@ -1078,14 +1095,13 @@ private void chainFuture(
}
@Override
- public String toString() {
- return "OffsetFetchRequestState{" +
- "requestedPartitions=" + requestedPartitions +
- ", memberId=" + memberInfo.memberId.orElse("undefined") +
- ", memberEpoch=" + (memberInfo.memberEpoch.isPresent() ? memberInfo.memberEpoch.get() : "undefined") +
- ", future=" + future +
- ", " + toStringBase() +
- '}';
+ public String toStringBase() {
+ return super.toStringBase() +
+ ", memberInfo=" + memberInfo +
+ ", expirationTimeMs=" + (expirationTimeMs().isPresent() ? expirationTimeMs() : "undefined") +
+ ", isExpired=" + isExpired +
+ ", requestedPartitions=" + requestedPartitions +
+ ", future=" + future;
}
}
@@ -1133,10 +1149,9 @@ private CompletableFuture> addOffsetFetch
inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
if (dupe.isPresent() || inflight.isPresent()) {
- log.debug("Duplicated unsent offset fetch request found for partitions: {}", request.requestedPartitions);
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
dupe.orElseGet(inflight::get).chainFuture(request.future);
} else {
- log.debug("Enqueuing offset fetch request for partitions: {}", request.requestedPartitions);
this.unsentOffsetFetches.add(request);
}
return request.future;
@@ -1154,7 +1169,7 @@ List drain(final long currentTimeMs) {
.filter(request -> !request.canSendRequest(currentTimeMs))
.collect(Collectors.toList());
- failAndRemoveExpiredCommitRequests();
+ failAndRemoveExpiredCommitRequests(currentTimeMs);
// Add all unsent offset commit requests to the unsentRequests list
List unsentRequests = unsentOffsetCommits.stream()
@@ -1168,7 +1183,7 @@ List drain(final long currentTimeMs) {
unsentOffsetFetches.stream()
.collect(Collectors.partitioningBy(request -> request.canSendRequest(currentTimeMs)));
- failAndRemoveExpiredFetchRequests();
+ failAndRemoveExpiredFetchRequests(currentTimeMs);
// Add all sendable offset fetch requests to the unsentRequests list and to the inflightOffsetFetches list
for (OffsetFetchRequestState request : partitionedBySendability.get(true)) {
@@ -1189,18 +1204,18 @@ List drain(final long currentTimeMs) {
* Find the unsent commit requests that have expired, remove them and complete their
* futures with a TimeoutException.
*/
- private void failAndRemoveExpiredCommitRequests() {
+ private void failAndRemoveExpiredCommitRequests(final long currentTimeMs) {
Queue requestsToPurge = new LinkedList<>(unsentOffsetCommits);
- requestsToPurge.forEach(RetriableRequestState::maybeExpire);
+ requestsToPurge.forEach(req -> req.maybeExpire(currentTimeMs));
}
/**
* Find the unsent fetch requests that have expired, remove them and complete their
* futures with a TimeoutException.
*/
- private void failAndRemoveExpiredFetchRequests() {
+ private void failAndRemoveExpiredFetchRequests(final long currentTimeMs) {
Queue requestsToPurge = new LinkedList<>(unsentOffsetFetches);
- requestsToPurge.forEach(RetriableRequestState::maybeExpire);
+ requestsToPurge.forEach(req -> req.maybeExpire(currentTimeMs));
}
private void clearAll() {
@@ -1278,5 +1293,11 @@ static class MemberInfo {
this.memberId = Optional.empty();
this.memberEpoch = Optional.empty();
}
+
+ @Override
+ public String toString() {
+ return "MemberInfo{" + "memberId=" + memberId.orElse("undefined") +
+ ", memberEpoch=" + (memberEpoch.isPresent() ? memberEpoch : "undefined") + "}";
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
index 81c45aba69c75..bd95e06c86448 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
@@ -91,7 +91,8 @@ public ConsumerDelegate create(LogContext logContext,
valueDeserializer,
client,
subscriptions,
- metadata
+ metadata,
+ assignors
);
else
return new LegacyKafkaConsumer<>(
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
index 7616ac6912289..adee6594603bb 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
@@ -293,13 +293,12 @@ private void closeInternal(final Duration timeout) {
* Check the unsent queue one last time and poll until all requests are sent or the timer runs out.
*/
private void sendUnsentRequests(final Timer timer) {
- if (!networkClientDelegate.hasAnyPendingRequests())
+ if (networkClientDelegate.unsentRequests().isEmpty())
return;
-
do {
networkClientDelegate.poll(timer.remainingMs(), timer.currentTimeMs());
timer.update();
- } while (timer.notExpired() && networkClientDelegate.hasAnyPendingRequests());
+ } while (timer.notExpired() && !networkClientDelegate.unsentRequests().isEmpty());
}
void cleanup() {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
index aba539b876326..a6cc28fb0f4c6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
@@ -28,6 +28,7 @@
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import java.util.Objects;
@@ -50,6 +51,7 @@
*/
public class CoordinatorRequestManager implements RequestManager {
private static final long COORDINATOR_DISCONNECT_LOGGING_INTERVAL_MS = 60 * 1000;
+ private final Time time;
private final Logger log;
private final BackgroundEventHandler backgroundEventHandler;
private final String groupId;
@@ -60,6 +62,7 @@ public class CoordinatorRequestManager implements RequestManager {
private Node coordinator;
public CoordinatorRequestManager(
+ final Time time,
final LogContext logContext,
final long retryBackoffMs,
final long retryBackoffMaxMs,
@@ -67,6 +70,7 @@ public CoordinatorRequestManager(
final String groupId
) {
Objects.requireNonNull(groupId);
+ this.time = time;
this.log = logContext.logger(this.getClass());
this.backgroundEventHandler = errorHandler;
this.groupId = groupId;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
index 0ed4a67b5869d..d31d412c65503 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
@@ -569,7 +569,7 @@ public ConsumerGroupHeartbeatRequestData buildRequestData() {
sentFields.rebalanceTimeoutMs = rebalanceTimeoutMs;
}
- // SubscribedTopicNames - only sent if it has changed since the last heartbeat
+ // SubscribedTopicNames - only sent if has changed since the last heartbeat
TreeSet subscribedTopicNames = new TreeSet<>(this.subscriptions.subscription());
if (sendAllFields || !subscribedTopicNames.equals(sentFields.subscribedTopicNames)) {
data.setSubscribedTopicNames(new ArrayList<>(this.subscriptions.subscription()));
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberState.java
index a62c634ba437e..9f0c7d947ea7e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberState.java
@@ -21,7 +21,6 @@
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
public enum MemberState {
@@ -121,7 +120,7 @@ public enum MemberState {
RECONCILING.previousValidStates = Arrays.asList(STABLE, JOINING, ACKNOWLEDGING, RECONCILING);
- ACKNOWLEDGING.previousValidStates = Collections.singletonList(RECONCILING);
+ ACKNOWLEDGING.previousValidStates = Arrays.asList(RECONCILING);
FATAL.previousValidStates = Arrays.asList(JOINING, STABLE, RECONCILING, ACKNOWLEDGING,
PREPARE_LEAVING, LEAVING, UNSUBSCRIBED);
@@ -134,11 +133,11 @@ public enum MemberState {
PREPARE_LEAVING.previousValidStates = Arrays.asList(JOINING, STABLE, RECONCILING,
ACKNOWLEDGING, UNSUBSCRIBED);
- LEAVING.previousValidStates = Collections.singletonList(PREPARE_LEAVING);
+ LEAVING.previousValidStates = Arrays.asList(PREPARE_LEAVING);
UNSUBSCRIBED.previousValidStates = Arrays.asList(PREPARE_LEAVING, LEAVING, FENCED);
- STALE.previousValidStates = Collections.singletonList(LEAVING);
+ STALE.previousValidStates = Arrays.asList(LEAVING);
}
private List previousValidStates;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
index 2aabf4ae130e8..76a550ad71985 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
@@ -960,7 +960,7 @@ void maybeReconcile() {
// best effort to commit the offsets in the case where the epoch might have changed while
// the current reconciliation is in process. Note this is using the rebalance timeout as
// it is the limit enforced by the broker to complete the reconciliation process.
- commitResult = commitRequestManager.maybeAutoCommitSyncBeforeRevocation(getDeadlineMsForTimeout(rebalanceTimeoutMs));
+ commitResult = commitRequestManager.maybeAutoCommitSyncBeforeRevocation(getExpirationTimeForTimeout(rebalanceTimeoutMs));
// Execute commit -> onPartitionsRevoked -> onPartitionsAssigned.
commitResult.whenComplete((__, commitReqError) -> {
@@ -986,7 +986,7 @@ void maybeReconcile() {
});
}
- long getDeadlineMsForTimeout(final long timeoutMs) {
+ long getExpirationTimeForTimeout(final long timeoutMs) {
long expiration = time.milliseconds() + timeoutMs;
if (expiration < 0) {
return Long.MAX_VALUE;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
index 310e0c417f617..141f5f955c8b5 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
@@ -21,12 +21,9 @@
import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.ClientUtils;
import org.apache.kafka.clients.KafkaClient;
-import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.NetworkClientUtils;
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
-import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.DisconnectException;
@@ -60,8 +57,6 @@
public class NetworkClientDelegate implements AutoCloseable {
private final KafkaClient client;
- private final BackgroundEventHandler backgroundEventHandler;
- private final Metadata metadata;
private final Time time;
private final Logger log;
private final int requestTimeoutMs;
@@ -72,13 +67,9 @@ public NetworkClientDelegate(
final Time time,
final ConsumerConfig config,
final LogContext logContext,
- final KafkaClient client,
- final Metadata metadata,
- final BackgroundEventHandler backgroundEventHandler) {
+ final KafkaClient client) {
this.time = time;
this.client = client;
- this.metadata = metadata;
- this.backgroundEventHandler = backgroundEventHandler;
this.log = logContext.logger(getClass());
this.unsentRequests = new ArrayDeque<>();
this.requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
@@ -136,25 +127,9 @@ public void poll(final long timeoutMs, final long currentTimeMs) {
pollTimeoutMs = Math.min(retryBackoffMs, pollTimeoutMs);
}
this.client.poll(pollTimeoutMs, currentTimeMs);
- maybePropagateMetadataError();
checkDisconnects(currentTimeMs);
}
- private void maybePropagateMetadataError() {
- try {
- metadata.maybeThrowAnyException();
- } catch (Exception e) {
- backgroundEventHandler.add(new ErrorEvent(e));
- }
- }
-
- /**
- * Return true if there is at least one in-flight request or unsent request.
- */
- public boolean hasAnyPendingRequests() {
- return client.hasInFlightRequests() || !unsentRequests.isEmpty();
- }
-
/**
* Tries to send the requests in the unsentRequest queue. If the request doesn't have an assigned node, it will
* find the leastLoadedOne, and will be retried in the next {@code poll()}. If the request is expired, a
@@ -334,20 +309,11 @@ Optional node() {
@Override
public String toString() {
- String remainingMs;
-
- if (timer != null) {
- timer.update();
- remainingMs = String.valueOf(timer.remainingMs());
- } else {
- remainingMs = "";
- }
-
return "UnsentRequest{" +
"requestBuilder=" + requestBuilder +
", handler=" + handler +
", node=" + node +
- ", remainingMs=" + remainingMs +
+ ", timer=" + timer +
'}';
}
}
@@ -405,8 +371,7 @@ public static Supplier supplier(final Time time,
final ApiVersions apiVersions,
final Metrics metrics,
final FetchMetricsManager fetchMetricsManager,
- final ClientTelemetrySender clientTelemetrySender,
- final BackgroundEventHandler backgroundEventHandler) {
+ final ClientTelemetrySender clientTelemetrySender) {
return new CachedSupplier() {
@Override
protected NetworkClientDelegate create() {
@@ -420,7 +385,7 @@ protected NetworkClientDelegate create() {
metadata,
fetchMetricsManager.throttleTimeSensor(),
clientTelemetrySender);
- return new NetworkClientDelegate(time, config, logContext, client, metadata, backgroundEventHandler);
+ return new NetworkClientDelegate(time, config, logContext, client);
}
};
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index 874631db84ec7..75d87432db680 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -155,7 +155,6 @@ protected RequestManagers create() {
apiVersions);
final TopicMetadataRequestManager topic = new TopicMetadataRequestManager(
logContext,
- time,
config);
HeartbeatRequestManager heartbeatRequestManager = null;
MembershipManager membershipManager = null;
@@ -164,7 +163,7 @@ protected RequestManagers create() {
if (groupRebalanceConfig != null && groupRebalanceConfig.groupId != null) {
Optional serverAssignor = Optional.ofNullable(config.getString(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG));
- coordinator = new CoordinatorRequestManager(
+ coordinator = new CoordinatorRequestManager(time,
logContext,
retryBackoffMs,
retryBackoffMaxMs,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerDelegate.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerDelegate.java
deleted file mode 100644
index 65dd9d2c30f91..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerDelegate.java
+++ /dev/null
@@ -1,37 +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.clients.consumer.internals;
-
-import org.apache.kafka.clients.consumer.ShareConsumer;
-import org.apache.kafka.common.metrics.Metrics;
-
-/**
- * This extension interface provides a handful of methods to expose internals of the {@link ShareConsumer} for
- * various tests.
- *
- *
- *
- * Note: this is for internal use only and is not intended for use by end users. Internal users should
- * not attempt to determine the underlying implementation to avoid coding to an unstable interface. Rather, it is
- * the {@link ShareConsumer} API contract that should serve as the caller's interface.
- */
-public interface ShareConsumerDelegate extends ShareConsumer {
-
- String clientId();
-
- Metrics metricsRegistry();
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerDelegateCreator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerDelegateCreator.java
deleted file mode 100644
index 3c41a58e0570b..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerDelegateCreator.java
+++ /dev/null
@@ -1,57 +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.clients.consumer.internals;
-
-import org.apache.kafka.clients.KafkaClient;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.KafkaShareConsumer;
-import org.apache.kafka.clients.consumer.ShareConsumer;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.utils.LogContext;
-import org.apache.kafka.common.utils.Time;
-
-/**
- * {@code ShareConsumerDelegateCreator} implements a quasi-factory pattern to allow the caller to remain unaware of the
- * underlying {@link ShareConsumer} implementation that is created. This provides the means by which
- * {@link KafkaShareConsumer} can remain the top-level facade for implementations, but allow different implementations
- * to co-exist under the covers.
- *
- *
- * Note: this is for internal use only and is not intended for use by end users. Internal users should
- * not attempt to determine the underlying implementation to avoid coding to an unstable interface. Rather, it is
- * the {@link ShareConsumer} API contract that should serve as the caller's interface.
- */
-public class ShareConsumerDelegateCreator {
- public ShareConsumerDelegate create(final ConsumerConfig config,
- final Deserializer keyDeserializer,
- final Deserializer valueDeserializer) {
- throw new UnsupportedOperationException("Not implemented");
- }
-
- public ShareConsumerDelegate create(final LogContext logContext,
- final String clientId,
- final String groupId,
- final ConsumerConfig config,
- final Deserializer keyDeserializer,
- final Deserializer valueDeserializer,
- final Time time,
- final KafkaClient client,
- final SubscriptionState subscriptions,
- final ConsumerMetadata metadata) {
- throw new UnsupportedOperationException("Not implemented");
- }
-}
\ No newline at end of file
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TimedRequestState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TimedRequestState.java
deleted file mode 100644
index c61032cea7249..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TimedRequestState.java
+++ /dev/null
@@ -1,71 +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.clients.consumer.internals;
-
-import org.apache.kafka.common.utils.LogContext;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.common.utils.Timer;
-
-/**
- * {@code TimedRequestState} adds to a {@link RequestState} a {@link Timer} with which to keep track
- * of the request's expiration.
- */
-public class TimedRequestState extends RequestState {
-
- private final Timer timer;
-
- public TimedRequestState(final LogContext logContext,
- final String owner,
- final long retryBackoffMs,
- final long retryBackoffMaxMs,
- final Timer timer) {
- super(logContext, owner, retryBackoffMs, retryBackoffMaxMs);
- this.timer = timer;
- }
-
- public TimedRequestState(final LogContext logContext,
- final String owner,
- final long retryBackoffMs,
- final int retryBackoffExpBase,
- final long retryBackoffMaxMs,
- final double jitter,
- final Timer timer) {
- super(logContext, owner, retryBackoffMs, retryBackoffExpBase, retryBackoffMaxMs, jitter);
- this.timer = timer;
- }
-
- public boolean isExpired() {
- timer.update();
- return timer.isExpired();
- }
-
- public long remainingMs() {
- timer.update();
- return timer.remainingMs();
- }
-
- public static Timer deadlineTimer(final Time time, final long deadlineMs) {
- long diff = Math.max(0, deadlineMs - time.milliseconds());
- return time.timer(diff);
- }
-
-
- @Override
- protected String toStringBase() {
- return super.toStringBase() + ", remainingMs=" + remainingMs();
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java
index a555d6ce7f359..75a5ed08d1554 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java
@@ -29,7 +29,6 @@
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.utils.LogContext;
-import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import java.util.Collections;
@@ -62,7 +61,6 @@
*/
public class TopicMetadataRequestManager implements RequestManager {
- private final Time time;
private final boolean allowAutoTopicCreation;
private final List inflightRequests;
private final long retryBackoffMs;
@@ -70,10 +68,9 @@ public class TopicMetadataRequestManager implements RequestManager {
private final Logger log;
private final LogContext logContext;
- public TopicMetadataRequestManager(final LogContext context, final Time time, final ConsumerConfig config) {
+ public TopicMetadataRequestManager(final LogContext context, final ConsumerConfig config) {
logContext = context;
log = logContext.logger(getClass());
- this.time = time;
inflightRequests = new LinkedList<>();
retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
@@ -84,7 +81,7 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
List expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
+ .filter(req -> req.isExpired(currentTimeMs))
.collect(Collectors.toList());
expiredRequests.forEach(TopicMetadataRequestState::expire);
@@ -102,10 +99,10 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
*
* @return the future of the metadata request.
*/
- public CompletableFuture>> requestAllTopicsMetadata(final long deadlineMs) {
+ public CompletableFuture>> requestAllTopicsMetadata(final long expirationTimeMs) {
TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
logContext,
- deadlineMs,
+ expirationTimeMs,
retryBackoffMs,
retryBackoffMaxMs);
inflightRequests.add(newRequest);
@@ -118,11 +115,11 @@ public CompletableFuture>> requestAllTopicsMetad
* @param topic to be requested.
* @return the future of the metadata request.
*/
- public CompletableFuture>> requestTopicMetadata(final String topic, final long deadlineMs) {
+ public CompletableFuture>> requestTopicMetadata(final String topic, final long expirationTimeMs) {
TopicMetadataRequestState newRequest = new TopicMetadataRequestState(
logContext,
topic,
- deadlineMs,
+ expirationTimeMs,
retryBackoffMs,
retryBackoffMaxMs);
inflightRequests.add(newRequest);
@@ -134,32 +131,35 @@ List inflightRequests() {
return inflightRequests;
}
- class TopicMetadataRequestState extends TimedRequestState {
+ class TopicMetadataRequestState extends RequestState {
private final String topic;
private final boolean allTopics;
+ private final long expirationTimeMs;
CompletableFuture>> future;
public TopicMetadataRequestState(final LogContext logContext,
- final long deadlineMs,
+ final long expirationTimeMs,
final long retryBackoffMs,
final long retryBackoffMaxMs) {
super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
- retryBackoffMaxMs, deadlineTimer(time, deadlineMs));
+ retryBackoffMaxMs);
future = new CompletableFuture<>();
this.topic = null;
this.allTopics = true;
+ this.expirationTimeMs = expirationTimeMs;
}
public TopicMetadataRequestState(final LogContext logContext,
final String topic,
- final long deadlineMs,
+ final long expirationTimeMs,
final long retryBackoffMs,
final long retryBackoffMaxMs) {
super(logContext, TopicMetadataRequestState.class.getSimpleName(), retryBackoffMs,
- retryBackoffMaxMs, deadlineTimer(time, deadlineMs));
+ retryBackoffMaxMs);
future = new CompletableFuture<>();
this.topic = topic;
this.allTopics = false;
+ this.expirationTimeMs = expirationTimeMs;
}
/**
@@ -167,6 +167,10 @@ public TopicMetadataRequestState(final LogContext logContext,
* {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest} if needed.
*/
private Optional send(final long currentTimeMs) {
+ if (currentTimeMs >= expirationTimeMs) {
+ return Optional.empty();
+ }
+
if (!canSendRequest(currentTimeMs)) {
return Optional.empty();
}
@@ -179,6 +183,10 @@ private Optional send(final long currentTim
return Optional.of(createUnsentRequest(request));
}
+ private boolean isExpired(final long currentTimeMs) {
+ return currentTimeMs >= expirationTimeMs;
+ }
+
private void expire() {
completeFutureAndRemoveRequest(
new TimeoutException("Timeout expired while fetching topic metadata"));
@@ -202,8 +210,9 @@ private NetworkClientDelegate.UnsentRequest createUnsentRequest(
private void handleError(final Throwable exception,
final long completionTimeMs) {
if (exception instanceof RetriableException) {
- if (isExpired()) {
- completeFutureAndRemoveRequest(new TimeoutException("Timeout expired while fetching topic metadata"));
+ if (completionTimeMs >= expirationTimeMs) {
+ completeFutureAndRemoveRequest(
+ new TimeoutException("Timeout expired while fetching topic metadata"));
} else {
onFailedAttempt(completionTimeMs);
}
@@ -213,12 +222,20 @@ private void handleError(final Throwable exception,
}
private void handleResponse(final ClientResponse response) {
+ long responseTimeMs = response.receivedTimeMs();
try {
Map> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
future.complete(res);
inflightRequests.remove(this);
- } catch (Exception e) {
- handleError(e, response.receivedTimeMs());
+ } catch (RetriableException e) {
+ if (responseTimeMs >= expirationTimeMs) {
+ completeFutureAndRemoveRequest(
+ new TimeoutException("Timeout expired while fetching topic metadata"));
+ } else {
+ onFailedAttempt(responseTimeMs);
+ }
+ } catch (Exception t) {
+ completeFutureAndRemoveRequest(t);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java
index 8cd17d19feb66..dffac12902177 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableApplicationEvent.java
@@ -43,7 +43,6 @@ public CompletableFuture future() {
return future;
}
- @Override
public long deadlineMs() {
return deadlineMs;
}
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 9ea7a32f949c5..12e77b0d516b7 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
@@ -855,12 +855,13 @@ private boolean shouldStopDrainBatchesForPartition(ProducerBatch first, TopicPar
}
int firstInFlightSequence = transactionManager.firstInFlightSequence(first.topicPartition);
- // If the queued batch already has an assigned sequence, then it is being retried.
- // In this case, we wait until the next immediate batch is ready and drain that.
- // We only move on when the next in line batch is complete (either successfully or due to
- // a fatal broker error). This effectively reduces our in flight request count to 1.
- return firstInFlightSequence != RecordBatch.NO_SEQUENCE && first.hasSequence()
- && first.baseSequence() != firstInFlightSequence;
+ if (firstInFlightSequence != RecordBatch.NO_SEQUENCE && first.hasSequence()
+ && first.baseSequence() != firstInFlightSequence)
+ // If the queued batch already has an assigned sequence, then it is being retried.
+ // In this case, we wait until the next immediate batch is ready and drain that.
+ // We only move on when the next in line batch is complete (either successfully or due to
+ // a fatal broker error). This effectively reduces our in flight request count to 1.
+ return true;
}
return false;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionMap.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionMap.java
index 3b3a711990822..ce8aa4a824f90 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionMap.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionMap.java
@@ -48,7 +48,7 @@ TxnPartitionEntry get(TopicPartition topicPartition) {
}
TxnPartitionEntry getOrCreate(TopicPartition topicPartition) {
- return topicPartitions.computeIfAbsent(topicPartition, TxnPartitionEntry::new);
+ return topicPartitions.computeIfAbsent(topicPartition, tp -> new TxnPartitionEntry(tp));
}
boolean contains(TopicPartition topicPartition) {
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 93f2f4225bc74..820adbdb5fbfa 100644
--- a/clients/src/main/java/org/apache/kafka/common/Cluster.java
+++ b/clients/src/main/java/org/apache/kafka/common/Cluster.java
@@ -246,8 +246,8 @@ public Node nodeById(int id) {
/**
* Get the node by node id if the replica for the given partition is online
- * @param partition The TopicPartition
- * @param id The node id
+ * @param partition
+ * @param id
* @return the node
*/
public Optional nodeIfOnline(TopicPartition partition, int id) {
diff --git a/clients/src/main/java/org/apache/kafka/common/ClusterResource.java b/clients/src/main/java/org/apache/kafka/common/ClusterResource.java
index 2f857ff560975..749f2d124e077 100644
--- a/clients/src/main/java/org/apache/kafka/common/ClusterResource.java
+++ b/clients/src/main/java/org/apache/kafka/common/ClusterResource.java
@@ -30,7 +30,7 @@ public class ClusterResource {
* Create {@link ClusterResource} with a cluster id. Note that cluster id may be {@code null} if the
* metadata request was sent to a broker without support for cluster ids. The first version of Kafka
* to support cluster id is 0.10.1.0.
- * @param clusterId The cluster id
+ * @param clusterId
*/
public ClusterResource(String clusterId) {
this.clusterId = clusterId;
diff --git a/clients/src/main/java/org/apache/kafka/common/ShareGroupState.java b/clients/src/main/java/org/apache/kafka/common/ShareGroupState.java
deleted file mode 100644
index 716421f3dea2a..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/ShareGroupState.java
+++ /dev/null
@@ -1,56 +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.util.Arrays;
-import java.util.Locale;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-/**
- * The share group state.
- */
-public enum ShareGroupState {
- UNKNOWN("Unknown"),
- STABLE("Stable"),
- DEAD("Dead"),
- EMPTY("Empty");
-
- private final static Map NAME_TO_ENUM = Arrays.stream(values())
- .collect(Collectors.toMap(state -> state.name.toUpperCase(Locale.ROOT), Function.identity()));
-
- private final String name;
-
- ShareGroupState(String name) {
- this.name = name;
- }
-
- /**
- * Case-insensitive share group state lookup by string name.
- */
- public static ShareGroupState parse(String name) {
- ShareGroupState state = NAME_TO_ENUM.get(name.toUpperCase(Locale.ROOT));
- return state == null ? UNKNOWN : state;
- }
-
- @Override
- public String toString() {
- return name;
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockInputStream.java b/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockInputStream.java
index 75576eedb5d75..e2c9ef8a63b6c 100644
--- a/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockInputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockInputStream.java
@@ -276,6 +276,11 @@ public void reset() {
throw new RuntimeException("reset not supported");
}
+ @Override
+ public boolean markSupported() {
+ return false;
+ }
+
/**
* Checks whether the version of lz4 on the classpath has the fix for reading from ByteBuffers with
* non-zero array offsets (see https://github.com/lz4/lz4-java/pull/65)
diff --git a/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java b/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java
index 0f1d213f50e36..97e370a383caf 100644
--- a/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java
@@ -333,6 +333,10 @@ public boolean isBlockChecksumSet() {
return blockChecksum == 1;
}
+ public boolean isBlockIndependenceSet() {
+ return blockIndependence == 1;
+ }
+
public int getVersion() {
return version;
}
diff --git a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
index f7632c8ca56df..b69ca06219a04 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
@@ -113,7 +113,9 @@ public AbstractConfig(ConfigDef definition, Map originals, Map
this.originals = resolveConfigVariables(configProviderProps, originalMap);
this.values = definition.parse(this.originals);
Map configUpdates = postProcessParsedConfig(Collections.unmodifiableMap(this.values));
- this.values.putAll(configUpdates);
+ for (Map.Entry update : configUpdates.entrySet()) {
+ this.values.put(update.getKey(), update.getValue());
+ }
definition.parse(this.values);
this.definition = definition;
if (doLog)
diff --git a/clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java b/clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java
index b0f1f1a29c78f..4fb6a483f66e7 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/internals/AllowedPaths.java
@@ -70,7 +70,7 @@ public Path parseUntrustedPath(String path) {
if (allowedPaths != null) {
Path normalisedPath = parsedPath.normalize();
- long allowed = allowedPaths.stream().filter(normalisedPath::startsWith).count();
+ long allowed = allowedPaths.stream().filter(allowedPath -> normalisedPath.startsWith(allowedPath)).count();
if (allowed == 0) {
return null;
}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/FencedStateEpochException.java b/clients/src/main/java/org/apache/kafka/common/errors/FencedStateEpochException.java
deleted file mode 100644
index 1e74bba199402..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/errors/FencedStateEpochException.java
+++ /dev/null
@@ -1,28 +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.errors;
-
-/**
- * Thrown when the share coordinator rejected the request because the share-group state epoch did not match.
- */
-public class FencedStateEpochException extends ApiException {
- private static final long serialVersionUID = 1L;
-
- public FencedStateEpochException(String message) {
- super(message);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidShareSessionEpochException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidShareSessionEpochException.java
deleted file mode 100644
index e261d8b7a8e88..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/errors/InvalidShareSessionEpochException.java
+++ /dev/null
@@ -1,28 +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.errors;
-
-/**
- * Thrown when the share session epoch is invalid.
- */
-public class InvalidShareSessionEpochException extends RetriableException {
- private static final long serialVersionUID = 1L;
-
- public InvalidShareSessionEpochException(String message) {
- super(message);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/ShareSessionNotFoundException.java b/clients/src/main/java/org/apache/kafka/common/errors/ShareSessionNotFoundException.java
deleted file mode 100644
index 2b2249f8a5831..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/errors/ShareSessionNotFoundException.java
+++ /dev/null
@@ -1,28 +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.errors;
-
-/**
- * Thrown when the share session was not found.
- */
-public class ShareSessionNotFoundException extends RetriableException {
- private static final long serialVersionUID = 1L;
-
- public ShareSessionNotFoundException(String message) {
- super(message);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetricsContext.java b/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetricsContext.java
index 5fafa144956b5..43eb8cb319465 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetricsContext.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetricsContext.java
@@ -21,7 +21,7 @@
import java.util.Map;
/**
- * An implementation of MetricsContext, it encapsulates required metrics context properties for Kafka services and clients
+ * A implementation of MetricsContext, it encapsulates required metrics context properties for Kafka services and clients
*/
public class KafkaMetricsContext implements MetricsContext {
/**
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
index 9e987aaedaed6..b52285dac63f4 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
@@ -392,7 +392,7 @@ public synchronized Sensor sensor(String name, MetricConfig config, Sensor.Recor
* receive every value recorded with this sensor.
* @param name The name of the sensor
* @param config A default configuration to use for this sensor for metrics that don't have their own config
- * @param inactiveSensorExpirationTimeSeconds If no value is recorded on the Sensor for this duration of time,
+ * @param inactiveSensorExpirationTimeSeconds If no value if recorded on the Sensor for this duration of time,
* it is eligible for removal
* @param parents The parent sensors
* @param recordingLevel The recording level.
@@ -419,7 +419,7 @@ public synchronized Sensor sensor(String name, MetricConfig config, long inactiv
* receive every value recorded with this sensor. This uses a default recording level of INFO.
* @param name The name of the sensor
* @param config A default configuration to use for this sensor for metrics that don't have their own config
- * @param inactiveSensorExpirationTimeSeconds If no value is recorded on the Sensor for this duration of time,
+ * @param inactiveSensorExpirationTimeSeconds If no value if recorded on the Sensor for this duration of time,
* it is eligible for removal
* @param parents The parent sensors
* @return The sensor that is created
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/MetricsReporter.java b/clients/src/main/java/org/apache/kafka/common/metrics/MetricsReporter.java
index e9b4582096dd1..75771fb4acedc 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/MetricsReporter.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/MetricsReporter.java
@@ -40,13 +40,13 @@ public interface MetricsReporter extends Reconfigurable, AutoCloseable {
/**
* This is called whenever a metric is updated or added
- * @param metric The metric that has been added or changed
+ * @param metric
*/
void metricChange(KafkaMetric metric);
/**
* This is called whenever a metric is removed
- * @param metric The metric that has been removed
+ * @param metric
*/
void metricRemoval(KafkaMetric metric);
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Histogram.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Histogram.java
index aa7de6fa6c931..97f91822d8261 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Histogram.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Histogram.java
@@ -116,6 +116,7 @@ public interface BinScheme {
public static class ConstantBinScheme implements BinScheme {
private static final int MIN_BIN_NUMBER = 0;
private final double min;
+ private final double max;
private final int bins;
private final double bucketWidth;
private final int maxBinNumber;
@@ -131,6 +132,7 @@ public ConstantBinScheme(int bins, double min, double max) {
if (bins < 2)
throw new IllegalArgumentException("Must have at least 2 bins.");
this.min = min;
+ this.max = max;
this.bins = bins;
this.bucketWidth = (max - min) / bins;
this.maxBinNumber = bins - 1;
diff --git a/clients/src/test/java/org/apache/kafka/common/network/DefaultChannelMetadataRegistry.java b/clients/src/main/java/org/apache/kafka/common/network/DefaultChannelMetadataRegistry.java
similarity index 93%
rename from clients/src/test/java/org/apache/kafka/common/network/DefaultChannelMetadataRegistry.java
rename to clients/src/main/java/org/apache/kafka/common/network/DefaultChannelMetadataRegistry.java
index 8985d00410c37..ae9e9a83a0c2c 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/DefaultChannelMetadataRegistry.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/DefaultChannelMetadataRegistry.java
@@ -22,7 +22,9 @@ public class DefaultChannelMetadataRegistry implements ChannelMetadataRegistry {
@Override
public void registerCipherInformation(final CipherInformation cipherInformation) {
- this.cipherInformation = cipherInformation;
+ if (this.cipherInformation != null) {
+ this.cipherInformation = cipherInformation;
+ }
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
index c6181b81c5e73..4ff41f81b594c 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
@@ -25,6 +25,7 @@
import org.apache.kafka.common.utils.Utils;
import java.io.Closeable;
+import java.io.IOException;
import java.net.InetAddress;
import java.nio.channels.SelectionKey;
import java.util.Map;
@@ -71,7 +72,7 @@ KafkaChannel buildChannel(String id, TransportLayer transportLayer, Supplier> APIS_BY_LISTENER =
new EnumMap<>(ApiMessageType.ListenerType.class);
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
index 10ae05aa850c9..900d191c8f9d4 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
@@ -41,7 +41,6 @@
import org.apache.kafka.common.errors.FencedInstanceIdException;
import org.apache.kafka.common.errors.FencedLeaderEpochException;
import org.apache.kafka.common.errors.FencedMemberEpochException;
-import org.apache.kafka.common.errors.FencedStateEpochException;
import org.apache.kafka.common.errors.FetchSessionIdNotFoundException;
import org.apache.kafka.common.errors.FetchSessionTopicIdException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
@@ -65,14 +64,12 @@
import org.apache.kafka.common.errors.InvalidPidMappingException;
import org.apache.kafka.common.errors.InvalidPrincipalTypeException;
import org.apache.kafka.common.errors.InvalidProducerEpochException;
-import org.apache.kafka.common.errors.InvalidRecordStateException;
import org.apache.kafka.common.errors.InvalidRegistrationException;
import org.apache.kafka.common.errors.InvalidReplicaAssignmentException;
import org.apache.kafka.common.errors.InvalidReplicationFactorException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.InvalidRequiredAcksException;
import org.apache.kafka.common.errors.InvalidSessionTimeoutException;
-import org.apache.kafka.common.errors.InvalidShareSessionEpochException;
import org.apache.kafka.common.errors.InvalidTimestampException;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.InvalidTxnStateException;
@@ -112,7 +109,6 @@
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.SaslAuthenticationException;
import org.apache.kafka.common.errors.SecurityDisabledException;
-import org.apache.kafka.common.errors.ShareSessionNotFoundException;
import org.apache.kafka.common.errors.SnapshotNotFoundException;
import org.apache.kafka.common.errors.StaleBrokerEpochException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
@@ -398,11 +394,7 @@ public enum Errors {
UNKNOWN_SUBSCRIPTION_ID(117, "Client sent a push telemetry request with an invalid or outdated subscription ID.", UnknownSubscriptionIdException::new),
TELEMETRY_TOO_LARGE(118, "Client sent a push telemetry request larger than the maximum size the broker will accept.", TelemetryTooLargeException::new),
INVALID_REGISTRATION(119, "The controller has considered the broker registration to be invalid.", InvalidRegistrationException::new),
- TRANSACTION_ABORTABLE(120, "The server encountered an error with the transaction. The client can abort the transaction to continue using this transactional ID.", TransactionAbortableException::new),
- INVALID_RECORD_STATE(121, "The record state is invalid. The acknowledgement of delivery could not be completed.", InvalidRecordStateException::new),
- SHARE_SESSION_NOT_FOUND(122, "The share session was not found.", ShareSessionNotFoundException::new),
- INVALID_SHARE_SESSION_EPOCH(123, "The share session epoch is invalid.", InvalidShareSessionEpochException::new),
- FENCED_STATE_EPOCH(124, "The share coordinator rejected the request because the share-group state epoch did not match.", FencedStateEpochException::new);
+ TRANSACTION_ABORTABLE(120, "The server encountered an error with the transaction. The client can abort the transaction to continue using this transactional ID.", TransactionAbortableException::new);
private static final Logger log = LoggerFactory.getLogger(Errors.class);
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/MessageSizeAccumulator.java b/clients/src/main/java/org/apache/kafka/common/protocol/MessageSizeAccumulator.java
index 98b6fc252a983..dac007ec5551f 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/MessageSizeAccumulator.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/MessageSizeAccumulator.java
@@ -40,6 +40,17 @@ public int sizeExcludingZeroCopy() {
return totalSize - zeroCopySize;
}
+ /**
+ * Get the total "zero-copy" size of the message. This is the summed
+ * total of all fields which have either have a type of 'bytes' with
+ * 'zeroCopy' enabled, or a type of 'records'
+ *
+ * @return total size of zero-copy data in the message
+ */
+ public int zeroCopySize() {
+ return zeroCopySize;
+ }
+
public void addZeroCopyBytes(int size) {
zeroCopySize += size;
totalSize += size;
diff --git a/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java
index 9ab8715236e74..0f29581623754 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/AbstractLegacyRecordBatch.java
@@ -279,7 +279,7 @@ static void writeHeader(DataOutputStream out, long offset, int size) throws IOEx
private static final class DataLogInputStream implements LogInputStream {
private final InputStream stream;
- private final int maxMessageSize;
+ protected final int maxMessageSize;
private final ByteBuffer offsetAndSizeBuffer;
DataLogInputStream(InputStream stream, int maxMessageSize) {
diff --git a/clients/src/main/java/org/apache/kafka/common/record/CompressionRatioEstimator.java b/clients/src/main/java/org/apache/kafka/common/record/CompressionRatioEstimator.java
index 0dbfb3e2f95b1..264525b380920 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/CompressionRatioEstimator.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/CompressionRatioEstimator.java
@@ -72,6 +72,13 @@ public static void resetEstimation(String topic) {
}
}
+ /**
+ * Remove the compression ratio estimation for a topic.
+ */
+ public static void removeEstimation(String topic) {
+ COMPRESSION_RATIO.remove(topic);
+ }
+
/**
* Set the compression estimation for a topic compression type combination. This method is for unit test purpose.
*/
diff --git a/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java b/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java
index 58fe9a0c9008a..014a5f4dfb135 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java
@@ -123,6 +123,14 @@ public boolean isValid() {
return sizeInBytes() >= RECORD_OVERHEAD_V0 && checksum() == computeChecksum();
}
+ public Long wrapperRecordTimestamp() {
+ return wrapperRecordTimestamp;
+ }
+
+ public TimestampType wrapperRecordTimestampType() {
+ return wrapperRecordTimestampType;
+ }
+
/**
* Throw an InvalidRecordException if isValid is false for this record
*/
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
index 70f279a6c29bc..a5985103ec0a0 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
@@ -80,10 +80,10 @@ public void write(int b) {
private long producerId;
private short producerEpoch;
private int baseSequence;
- private int uncompressedRecordsSizeInBytes; // Number of bytes (excluding the header) written before compression
- private int numRecords;
- private float actualCompressionRatio;
- private long maxTimestamp;
+ private int uncompressedRecordsSizeInBytes = 0; // Number of bytes (excluding the header) written before compression
+ private int numRecords = 0;
+ private float actualCompressionRatio = 1;
+ private long maxTimestamp = RecordBatch.NO_TIMESTAMP;
private long offsetOfMaxTimestamp = -1;
private Long lastOffset = null;
private Long baseTimestamp = null;
@@ -814,7 +814,7 @@ private void ensureOpenForRecordBatchWrite() {
}
/**
- * Get an estimate of the number of bytes written (based on the estimation factor hard-coded in {@link CompressionType}).
+ * Get an estimate of the number of bytes written (based on the estimation factor hard-coded in {@link CompressionType}.
* @return The estimated number of bytes written
*/
private int estimatedBytesWritten() {
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
index eaa5e6dcb8bbd..b51221f5af642 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
@@ -326,20 +326,6 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
return ListClientMetricsResourcesRequest.parse(buffer, apiVersion);
case DESCRIBE_TOPIC_PARTITIONS:
return DescribeTopicPartitionsRequest.parse(buffer, apiVersion);
- case SHARE_GROUP_HEARTBEAT:
- return ShareGroupHeartbeatRequest.parse(buffer, apiVersion);
- case SHARE_GROUP_DESCRIBE:
- return ShareGroupDescribeRequest.parse(buffer, apiVersion);
- case SHARE_FETCH:
- return ShareFetchRequest.parse(buffer, apiVersion);
- case SHARE_ACKNOWLEDGE:
- return ShareAcknowledgeRequest.parse(buffer, apiVersion);
- case ADD_RAFT_VOTER:
- return AddRaftVoterRequest.parse(buffer, apiVersion);
- case REMOVE_RAFT_VOTER:
- return RemoveRaftVoterRequest.parse(buffer, apiVersion);
- case UPDATE_RAFT_VOTER:
- return UpdateRaftVoterRequest.parse(buffer, apiVersion);
default:
throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
"code should be updated to do so.", apiKey));
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
index 8210fb2a7f6b3..dbafdbf3bcb07 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
@@ -263,20 +263,6 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, ByteBuffer response
return ListClientMetricsResourcesResponse.parse(responseBuffer, version);
case DESCRIBE_TOPIC_PARTITIONS:
return DescribeTopicPartitionsResponse.parse(responseBuffer, version);
- case SHARE_GROUP_HEARTBEAT:
- return ShareGroupHeartbeatResponse.parse(responseBuffer, version);
- case SHARE_GROUP_DESCRIBE:
- return ShareGroupDescribeResponse.parse(responseBuffer, version);
- case SHARE_FETCH:
- return ShareFetchResponse.parse(responseBuffer, version);
- case SHARE_ACKNOWLEDGE:
- return ShareAcknowledgeResponse.parse(responseBuffer, version);
- case ADD_RAFT_VOTER:
- return AddRaftVoterResponse.parse(responseBuffer, version);
- case REMOVE_RAFT_VOTER:
- return RemoveRaftVoterResponse.parse(responseBuffer, version);
- case UPDATE_RAFT_VOTER:
- return UpdateRaftVoterResponse.parse(responseBuffer, version);
default:
throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
"code should be updated to do so.", apiKey));
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterRequest.java
deleted file mode 100644
index 2d385d861c4ac..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterRequest.java
+++ /dev/null
@@ -1,75 +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.requests;
-
-import org.apache.kafka.common.message.AddRaftVoterRequestData;
-import org.apache.kafka.common.message.AddRaftVoterResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-
-public class AddRaftVoterRequest extends AbstractRequest {
- public static class Builder extends AbstractRequest.Builder {
- private final AddRaftVoterRequestData data;
-
- public Builder(AddRaftVoterRequestData data) {
- super(ApiKeys.ADD_RAFT_VOTER);
- this.data = data;
- }
-
- @Override
- public AddRaftVoterRequest build(short version) {
- return new AddRaftVoterRequest(data, version);
- }
-
- @Override
- public String toString() {
- return data.toString();
- }
-
- }
-
- private final AddRaftVoterRequestData data;
-
- public AddRaftVoterRequest(AddRaftVoterRequestData data, short version) {
- super(ApiKeys.ADD_RAFT_VOTER, version);
- this.data = data;
- }
-
- @Override
- public AddRaftVoterRequestData data() {
- return data;
- }
-
- @Override
- public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
- Errors error = Errors.forException(e);
- return new AddRaftVoterResponse(new AddRaftVoterResponseData().
- setErrorCode(error.code()).
- setErrorMessage(error.message()).
- setThrottleTimeMs(throttleTimeMs));
- }
-
- public static AddRaftVoterRequest parse(ByteBuffer buffer, short version) {
- return new AddRaftVoterRequest(
- new AddRaftVoterRequestData(new ByteBufferAccessor(buffer), version),
- version);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterResponse.java
deleted file mode 100644
index 3b931702d4419..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterResponse.java
+++ /dev/null
@@ -1,65 +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.requests;
-
-import org.apache.kafka.common.message.AddRaftVoterResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Map;
-
-public class AddRaftVoterResponse extends AbstractResponse {
- private final AddRaftVoterResponseData data;
-
- public AddRaftVoterResponse(AddRaftVoterResponseData data) {
- super(ApiKeys.ADD_RAFT_VOTER);
- this.data = data;
- }
-
- @Override
- public AddRaftVoterResponseData data() {
- return data;
- }
-
- @Override
- public int throttleTimeMs() {
- return data.throttleTimeMs();
- }
-
- @Override
- public void maybeSetThrottleTimeMs(int throttleTimeMs) {
- // not supported
- }
-
- @Override
- public Map errorCounts() {
- if (data.errorCode() != Errors.NONE.code()) {
- return Collections.singletonMap(Errors.forCode(data.errorCode()), 1);
- } else {
- return Collections.emptyMap();
- }
- }
-
- public static AddRaftVoterResponse parse(ByteBuffer buffer, short version) {
- return new AddRaftVoterResponse(
- new AddRaftVoterResponseData(new ByteBufferAccessor(buffer), version));
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java
index 142210f765d01..0aae47aa24a96 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java
@@ -83,20 +83,20 @@ public Builder(AlterPartitionRequestData data, boolean canUseTopicIds) {
@Override
public AlterPartitionRequest build(short version) {
if (version < 3) {
- data.topics().forEach(topicData ->
+ data.topics().forEach(topicData -> {
topicData.partitions().forEach(partitionData -> {
// The newIsrWithEpochs will be empty after build. Then we can skip the conversion if the build
// is called again.
if (!partitionData.newIsrWithEpochs().isEmpty()) {
List newIsr = new ArrayList<>(partitionData.newIsrWithEpochs().size());
- partitionData.newIsrWithEpochs().forEach(brokerState ->
- newIsr.add(brokerState.brokerId())
- );
+ partitionData.newIsrWithEpochs().forEach(brokerState -> {
+ newIsr.add(brokerState.brokerId());
+ });
partitionData.setNewIsr(newIsr);
partitionData.setNewIsrWithEpochs(Collections.emptyList());
}
- })
- );
+ });
+ });
}
return new AlterPartitionRequest(data, version);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionResponse.java
index 38b8eaf275bbd..9ee92f7b809cd 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionResponse.java
@@ -44,9 +44,9 @@ public AlterPartitionResponseData data() {
public Map errorCounts() {
Map counts = new HashMap<>();
updateErrorCounts(counts, Errors.forCode(data.errorCode()));
- data.topics().forEach(topicResponse -> topicResponse.partitions().forEach(partitionResponse ->
- updateErrorCounts(counts, Errors.forCode(partitionResponse.errorCode()))
- ));
+ data.topics().forEach(topicResponse -> topicResponse.partitions().forEach(partitionResponse -> {
+ updateErrorCounts(counts, Errors.forCode(partitionResponse.errorCode()));
+ }));
return counts;
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
index d3b606eeac63a..bc5aa0ba35a33 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
@@ -76,11 +76,11 @@ public static ControlledShutdownResponse prepareResponse(Errors error, Set
+ tps.forEach(tp -> {
pSet.add(new RemainingPartition()
.setTopicName(tp.topic())
- .setPartitionIndex(tp.partition()))
- );
+ .setPartitionIndex(tp.partition()));
+ });
data.setRemainingPartitions(pSet);
return new ControlledShutdownResponse(data);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
index 2cf8dd40a15c9..27e7c581c33bb 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
@@ -107,7 +107,7 @@ private void validate(short version) {
}
public static DeleteAclsFilterResult filterResult(AclDeleteResult result) {
- ApiError error = result.exception().map(ApiError::fromThrowable).orElse(ApiError.NONE);
+ ApiError error = result.exception().map(e -> ApiError.fromThrowable(e)).orElse(ApiError.NONE);
List matchingAcls = result.aclBindingDeleteResults().stream()
.map(DeleteAclsResponse::matchingAcl)
.collect(Collectors.toList());
@@ -118,7 +118,7 @@ public static DeleteAclsFilterResult filterResult(AclDeleteResult result) {
}
private static DeleteAclsMatchingAcl matchingAcl(AclDeleteResult.AclBindingDeleteResult result) {
- ApiError error = result.exception().map(ApiError::fromThrowable).orElse(ApiError.NONE);
+ ApiError error = result.exception().map(e -> ApiError.fromThrowable(e)).orElse(ApiError.NONE);
AclBinding acl = result.aclBinding();
return matchingAcl(acl, error);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
index 60b4031725e80..006491f74c067 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsRequest.java
@@ -92,7 +92,7 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
public List topicNames() {
if (version() >= 6)
- return data.topics().stream().map(DeleteTopicState::name).collect(Collectors.toList());
+ return data.topics().stream().map(topic -> topic.name()).collect(Collectors.toList());
return data.topicNames();
}
@@ -104,7 +104,7 @@ public int numberOfTopics() {
public List topicIds() {
if (version() >= 6)
- return data.topics().stream().map(DeleteTopicState::topicId).collect(Collectors.toList());
+ return data.topics().stream().map(topic -> topic.topicId()).collect(Collectors.toList());
return Collections.emptyList();
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsRequest.java
index 1bae21a9e9c91..d612ca80949a0 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsRequest.java
@@ -58,12 +58,12 @@ public DescribeConfigsResponse getErrorResponse(int throttleTimeMs, Throwable e)
Errors error = Errors.forException(e);
return new DescribeConfigsResponse(new DescribeConfigsResponseData()
.setThrottleTimeMs(throttleTimeMs)
- .setResults(data.resources().stream().map(result ->
- new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(error.code())
+ .setResults(data.resources().stream().map(result -> {
+ return new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(error.code())
.setErrorMessage(error.message())
.setResourceName(result.resourceName())
- .setResourceType(result.resourceType()))
- .collect(Collectors.toList())
+ .setResourceType(result.resourceType());
+ }).collect(Collectors.toList())
));
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
index e0bba392a076b..cbf3054217363 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
@@ -59,9 +59,9 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
public Map errorCounts() {
Map errorCounts = new HashMap<>();
errorCounts.put(Errors.forCode(data.errorCode()), 1);
- data.results().forEach(result ->
- updateErrorCounts(errorCounts, Errors.forCode(result.errorCode()))
- );
+ data.results().forEach(result -> {
+ updateErrorCounts(errorCounts, Errors.forCode(result.errorCode()));
+ });
return errorCounts;
}
@@ -93,11 +93,13 @@ public LogDirInfo(Errors error, Map replicaInfos) {
@Override
public String toString() {
- return "(error=" +
- error +
- ", replicas=" +
- replicaInfos +
- ")";
+ StringBuilder builder = new StringBuilder();
+ builder.append("(error=")
+ .append(error)
+ .append(", replicas=")
+ .append(replicaInfos)
+ .append(")");
+ return builder.toString();
}
}
@@ -124,13 +126,15 @@ public ReplicaInfo(long size, long offsetLag, boolean isFuture) {
@Override
public String toString() {
- return "(size=" +
- size +
- ", offsetLag=" +
- offsetLag +
- ", isFuture=" +
- isFuture +
- ")";
+ StringBuilder builder = new StringBuilder();
+ builder.append("(size=")
+ .append(size)
+ .append(", offsetLag=")
+ .append(offsetLag)
+ .append(", isFuture=")
+ .append(isFuture)
+ .append(")");
+ return builder.toString();
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
index 1082200ec393c..2065a15d94259 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
@@ -316,18 +316,20 @@ public FetchRequest build(short version) {
@Override
public String toString() {
- return "(type=FetchRequest" +
- ", replicaId=" + replicaId +
- ", maxWait=" + maxWait +
- ", minBytes=" + minBytes +
- ", maxBytes=" + maxBytes +
- ", fetchData=" + toFetch +
- ", isolationLevel=" + isolationLevel +
- ", removed=" + removed.stream().map(TopicIdPartition::toString).collect(Collectors.joining(", ")) +
- ", replaced=" + replaced.stream().map(TopicIdPartition::toString).collect(Collectors.joining(", ")) +
- ", metadata=" + metadata +
- ", rackId=" + rackId +
- ")";
+ StringBuilder bld = new StringBuilder();
+ bld.append("(type=FetchRequest").
+ append(", replicaId=").append(replicaId).
+ append(", maxWait=").append(maxWait).
+ append(", minBytes=").append(minBytes).
+ append(", maxBytes=").append(maxBytes).
+ append(", fetchData=").append(toFetch).
+ append(", isolationLevel=").append(isolationLevel).
+ append(", removed=").append(removed.stream().map(TopicIdPartition::toString).collect(Collectors.joining(", "))).
+ append(", replaced=").append(replaced.stream().map(TopicIdPartition::toString).collect(Collectors.joining(", "))).
+ append(", metadata=").append(metadata).
+ append(", rackId=").append(rackId).
+ append(")");
+ return bld.toString();
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchSnapshotResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchSnapshotResponse.java
index 2ea0265f6992d..c86ba9cfc6616 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchSnapshotResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchSnapshotResponse.java
@@ -84,7 +84,7 @@ public static FetchSnapshotResponseData withTopLevelError(Errors error) {
/**
* Creates a FetchSnapshotResponseData with a single PartitionSnapshot for the topic partition.
*
- * The partition index will already be populated when calling operator.
+ * The partition index will already by populated when calling operator.
*
* @param topicPartition the topic partition to include
* @param operator unary operator responsible for populating all of the appropriate fields
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java
index 10041633faeed..af10a6c78aebd 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java
@@ -86,7 +86,7 @@ public static String maybeTruncateReason(final String reason) {
/**
* Since JoinGroupRequest version 4, a client that sends a join group request with
- * {@link #UNKNOWN_MEMBER_ID} needs to rejoin with a new member id generated
+ * {@link UNKNOWN_MEMBER_ID} needs to rejoin with a new member id generated
* by the server. Once the second join group request is complete, the client is
* added as a new member of the group.
*
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java
index 9fc83cfd847a8..8caddb0054169 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/LeaderAndIsrRequest.java
@@ -112,14 +112,16 @@ private static Map groupByTopic(List errorCounts() {
updateErrorCounts(combinedErrorCounts, Errors.forCode(data.errorCode()));
// Member level error.
- data.members().forEach(memberResponse ->
- updateErrorCounts(combinedErrorCounts, Errors.forCode(memberResponse.errorCode()))
- );
+ data.members().forEach(memberResponse -> {
+ updateErrorCounts(combinedErrorCounts, Errors.forCode(memberResponse.errorCode()));
+ });
return combinedErrorCounts;
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java
index 57aeb9de1bbb5..fc996453d6470 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ListOffsetsRequest.java
@@ -181,4 +181,13 @@ public static List toListOffsetsTopics(Map(topics.values());
}
+
+ public static ListOffsetsTopic singletonRequestData(String topic, int partitionIndex, long timestamp, int maxNumOffsets) {
+ return new ListOffsetsTopic()
+ .setName(topic)
+ .setPartitions(Collections.singletonList(new ListOffsetsPartition()
+ .setPartitionIndex(partitionIndex)
+ .setTimestamp(timestamp)
+ .setMaxNumOffsets(maxNumOffsets)));
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
index 88111b1007717..0eba5f29ab4a6 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java
@@ -101,11 +101,11 @@ public static OffsetCommitResponseData getErrorResponse(
.setName(topic.name());
response.topics().add(responseTopic);
- topic.partitions().forEach(partition ->
+ topic.partitions().forEach(partition -> {
responseTopic.partitions().add(new OffsetCommitResponsePartition()
.setPartitionIndex(partition.partitionIndex())
- .setErrorCode(error.code()))
- );
+ .setErrorCode(error.code()));
+ });
});
return response;
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
index 2b6d00b1a47f6..9848ad380fab0 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
@@ -160,11 +160,13 @@ public Builder addPartitions(
Errors error
) {
final OffsetCommitResponseTopic topicResponse = getOrCreateTopic(topicName);
- partitions.forEach(partition ->
+
+ partitions.forEach(partition -> {
topicResponse.partitions().add(new OffsetCommitResponsePartition()
.setPartitionIndex(partitionIndex.apply(partition))
- .setErrorCode(error.code()))
- );
+ .setErrorCode(error.code()));
+ });
+
return this;
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java
index aa9b4bc4ffe66..e8c21969e9e3b 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java
@@ -83,11 +83,13 @@ public
Builder addPartitions(
Errors error
) {
final OffsetDeleteResponseTopic topicResponse = getOrCreateTopic(topicName);
- partitions.forEach(partition ->
+
+ partitions.forEach(partition -> {
topicResponse.partitions().add(new OffsetDeleteResponsePartition()
.setPartitionIndex(partitionIndex.apply(partition))
- .setErrorCode(error.code()))
- );
+ .setErrorCode(error.code()));
+ });
+
return this;
}
@@ -111,9 +113,9 @@ public Builder merge(
// Otherwise, we add the partitions to the existing one. Note we
// expect non-overlapping partitions here as we don't verify
// if the partition is already in the list before adding it.
- newTopic.partitions().forEach(partition ->
- existingTopic.partitions().add(partition.duplicate())
- );
+ newTopic.partitions().forEach(partition -> {
+ existingTopic.partitions().add(partition.duplicate());
+ });
}
});
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
index 1b748dca28e61..70473565f63d5 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java
@@ -237,12 +237,12 @@ public List groups() {
group.setTopics(null);
} else {
// Otherwise, topics are translated to the new structure.
- data.topics().forEach(topic ->
+ data.topics().forEach(topic -> {
group.topics().add(new OffsetFetchRequestTopics()
.setName(topic.name())
.setPartitionIndexes(topic.partitionIndexes())
- )
- );
+ );
+ });
}
return Collections.singletonList(group);
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
index 3f2679c95aa72..4724ce4789ccb 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
@@ -94,12 +94,14 @@ private ProduceRequest build(short version, boolean validate) {
@Override
public String toString() {
- return "(type=ProduceRequest" +
- ", acks=" + data.acks() +
- ", timeout=" + data.timeoutMs() +
- ", partitionRecords=(" + data.topicData().stream().flatMap(d -> d.partitionData().stream()).collect(Collectors.toList()) +
- "), transactionalId='" + (data.transactionalId() != null ? data.transactionalId() : "") +
- "'";
+ StringBuilder bld = new StringBuilder();
+ bld.append("(type=ProduceRequest")
+ .append(", acks=").append(data.acks())
+ .append(", timeout=").append(data.timeoutMs())
+ .append(", partitionRecords=(").append(data.topicData().stream().flatMap(d -> d.partitionData().stream()).collect(Collectors.toList()))
+ .append("), transactionalId='").append(data.transactionalId() != null ? data.transactionalId() : "")
+ .append("'");
+ return bld.toString();
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RemoveRaftVoterRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/RemoveRaftVoterRequest.java
deleted file mode 100644
index cf5f1dc0ce20b..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/RemoveRaftVoterRequest.java
+++ /dev/null
@@ -1,75 +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.requests;
-
-import org.apache.kafka.common.message.RemoveRaftVoterRequestData;
-import org.apache.kafka.common.message.RemoveRaftVoterResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-
-public class RemoveRaftVoterRequest extends AbstractRequest {
- public static class Builder extends AbstractRequest.Builder {
- private final RemoveRaftVoterRequestData data;
-
- public Builder(RemoveRaftVoterRequestData data) {
- super(ApiKeys.REMOVE_RAFT_VOTER);
- this.data = data;
- }
-
- @Override
- public RemoveRaftVoterRequest build(short version) {
- return new RemoveRaftVoterRequest(data, version);
- }
-
- @Override
- public String toString() {
- return data.toString();
- }
-
- }
-
- private final RemoveRaftVoterRequestData data;
-
- public RemoveRaftVoterRequest(RemoveRaftVoterRequestData data, short version) {
- super(ApiKeys.REMOVE_RAFT_VOTER, version);
- this.data = data;
- }
-
- @Override
- public RemoveRaftVoterRequestData data() {
- return data;
- }
-
- @Override
- public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
- Errors error = Errors.forException(e);
- return new RemoveRaftVoterResponse(new RemoveRaftVoterResponseData().
- setErrorCode(error.code()).
- setErrorMessage(error.message()).
- setThrottleTimeMs(throttleTimeMs));
- }
-
- public static RemoveRaftVoterRequest parse(ByteBuffer buffer, short version) {
- return new RemoveRaftVoterRequest(
- new RemoveRaftVoterRequestData(new ByteBufferAccessor(buffer), version),
- version);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RemoveRaftVoterResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/RemoveRaftVoterResponse.java
deleted file mode 100644
index a74711e2f836f..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/RemoveRaftVoterResponse.java
+++ /dev/null
@@ -1,65 +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.requests;
-
-import org.apache.kafka.common.message.RemoveRaftVoterResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Map;
-
-public class RemoveRaftVoterResponse extends AbstractResponse {
- private final RemoveRaftVoterResponseData data;
-
- public RemoveRaftVoterResponse(RemoveRaftVoterResponseData data) {
- super(ApiKeys.REMOVE_RAFT_VOTER);
- this.data = data;
- }
-
- @Override
- public RemoveRaftVoterResponseData data() {
- return data;
- }
-
- @Override
- public int throttleTimeMs() {
- return data.throttleTimeMs();
- }
-
- @Override
- public void maybeSetThrottleTimeMs(int throttleTimeMs) {
- // not supported
- }
-
- @Override
- public Map errorCounts() {
- if (data.errorCode() != Errors.NONE.code()) {
- return Collections.singletonMap(Errors.forCode(data.errorCode()), 1);
- } else {
- return Collections.emptyMap();
- }
- }
-
- public static RemoveRaftVoterResponse parse(ByteBuffer buffer, short version) {
- return new RemoveRaftVoterResponse(
- new RemoveRaftVoterResponseData(new ByteBufferAccessor(buffer), version));
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ShareAcknowledgeRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ShareAcknowledgeRequest.java
deleted file mode 100644
index 1b77b43be33c1..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/ShareAcknowledgeRequest.java
+++ /dev/null
@@ -1,127 +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.requests;
-
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.message.ShareAcknowledgeRequestData;
-import org.apache.kafka.common.message.ShareAcknowledgeResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class ShareAcknowledgeRequest extends AbstractRequest {
-
- public static class Builder extends AbstractRequest.Builder {
-
- private final ShareAcknowledgeRequestData data;
-
- public Builder(ShareAcknowledgeRequestData data) {
- this(data, false);
- }
-
- public Builder(ShareAcknowledgeRequestData data, boolean enableUnstableLastVersion) {
- super(ApiKeys.SHARE_ACKNOWLEDGE, enableUnstableLastVersion);
- this.data = data;
- }
-
- public static ShareAcknowledgeRequest.Builder forConsumer(String groupId, ShareFetchMetadata metadata,
- Map> acknowledgementsMap) {
- ShareAcknowledgeRequestData data = new ShareAcknowledgeRequestData();
- data.setGroupId(groupId);
- if (metadata != null) {
- data.setMemberId(metadata.memberId().toString());
- data.setShareSessionEpoch(metadata.epoch());
- }
-
- // Build a map of topics to acknowledge keyed by topic ID, and within each a map of partitions keyed by index
- Map> ackMap = new HashMap<>();
-
- for (Map.Entry