sessionTopicPartitions() {
+ return sessionPartitions.keySet();
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
index fdacc09db8c8a..1d469a66436a6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -36,6 +36,7 @@
import java.time.Duration;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -919,12 +920,21 @@ default ListConsumerGroupsResult listConsumerGroups() {
* @param options The options to use when listing the consumer group offsets.
* @return The ListGroupOffsetsResult
*/
- ListConsumerGroupOffsetsResult listConsumerGroupOffsets(String groupId, ListConsumerGroupOffsetsOptions options);
+ default ListConsumerGroupOffsetsResult listConsumerGroupOffsets(String groupId, ListConsumerGroupOffsetsOptions options) {
+ @SuppressWarnings("deprecation")
+ ListConsumerGroupOffsetsSpec groupSpec = new ListConsumerGroupOffsetsSpec()
+ .topicPartitions(options.topicPartitions());
+
+ // We can use the provided options with the batched API, which uses topic partitions from
+ // the group spec and ignores any topic partitions set in the options.
+ return listConsumerGroupOffsets(Collections.singletonMap(groupId, groupSpec), options);
+ }
/**
* List the consumer group offsets available in the cluster with the default options.
*
- * This is a convenience method for {@link #listConsumerGroupOffsets(String, ListConsumerGroupOffsetsOptions)} with default options.
+ * This is a convenience method for {@link #listConsumerGroupOffsets(Map, ListConsumerGroupOffsetsOptions)}
+ * to list offsets of all partitions of one group with default options.
*
* @return The ListGroupOffsetsResult.
*/
@@ -932,6 +942,29 @@ default ListConsumerGroupOffsetsResult listConsumerGroupOffsets(String groupId)
return listConsumerGroupOffsets(groupId, new ListConsumerGroupOffsetsOptions());
}
+ /**
+ * List the consumer group offsets available in the cluster for the specified consumer groups.
+ *
+ * @param groupSpecs Map of consumer group ids to a spec that specifies the topic partitions of the group to list offsets for.
+ *
+ * @param options The options to use when listing the consumer group offsets.
+ * @return The ListConsumerGroupOffsetsResult
+ */
+ ListConsumerGroupOffsetsResult listConsumerGroupOffsets(Map groupSpecs, ListConsumerGroupOffsetsOptions options);
+
+ /**
+ * List the consumer group offsets available in the cluster for the specified groups with the default options.
+ *
+ * This is a convenience method for
+ * {@link #listConsumerGroupOffsets(Map, ListConsumerGroupOffsetsOptions)} with default options.
+ *
+ * @param groupSpecs Map of consumer group ids to a spec that specifies the topic partitions of the group to list offsets for.
+ * @return The ListConsumerGroupOffsetsResult.
+ */
+ default ListConsumerGroupOffsetsResult listConsumerGroupOffsets(Map groupSpecs) {
+ return listConsumerGroupOffsets(groupSpecs, new ListConsumerGroupOffsetsOptions());
+ }
+
/**
* Delete consumer groups from the cluster.
*
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/FinalizedVersionRange.java b/clients/src/main/java/org/apache/kafka/clients/admin/FinalizedVersionRange.java
index aa0401a8a86eb..1442de58513f9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/FinalizedVersionRange.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/FinalizedVersionRange.java
@@ -36,10 +36,10 @@ public class FinalizedVersionRange {
* @throws IllegalArgumentException Raised when the condition described above is not met.
*/
FinalizedVersionRange(final short minVersionLevel, final short maxVersionLevel) {
- if (minVersionLevel < 1 || maxVersionLevel < 1 || maxVersionLevel < minVersionLevel) {
+ if (minVersionLevel < 0 || maxVersionLevel < 0 || maxVersionLevel < minVersionLevel) {
throw new IllegalArgumentException(
String.format(
- "Expected minVersionLevel >= 1, maxVersionLevel >= 1 and" +
+ "Expected minVersionLevel >= 0, maxVersionLevel >= 0 and" +
" maxVersionLevel >= minVersionLevel, but received" +
" minVersionLevel: %d, maxVersionLevel: %d", minVersionLevel, maxVersionLevel));
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index aad2610c94a45..89bc011a4d504 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -218,6 +218,7 @@
import org.apache.kafka.common.requests.ExpireDelegationTokenResponse;
import org.apache.kafka.common.requests.IncrementalAlterConfigsRequest;
import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
+import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.requests.ListGroupsRequest;
import org.apache.kafka.common.requests.ListGroupsResponse;
import org.apache.kafka.common.requests.ListOffsetsRequest;
@@ -273,8 +274,8 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.apache.kafka.common.internals.Topic.METADATA_TOPIC_NAME;
-import static org.apache.kafka.common.internals.Topic.METADATA_TOPIC_PARTITION;
+import static org.apache.kafka.common.internals.Topic.CLUSTER_METADATA_TOPIC_NAME;
+import static org.apache.kafka.common.internals.Topic.CLUSTER_METADATA_TOPIC_PARTITION;
import static org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData.ReassignablePartition;
import static org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData.ReassignablePartitionResponse;
import static org.apache.kafka.common.message.AlterPartitionReassignmentsResponseData.ReassignableTopicResponse;
@@ -3400,13 +3401,14 @@ void handleFailure(Throwable throwable) {
}
@Override
- public ListConsumerGroupOffsetsResult listConsumerGroupOffsets(final String groupId,
- final ListConsumerGroupOffsetsOptions options) {
+ public ListConsumerGroupOffsetsResult listConsumerGroupOffsets(Map groupSpecs,
+ ListConsumerGroupOffsetsOptions options) {
SimpleAdminApiFuture> future =
- ListConsumerGroupOffsetsHandler.newFuture(groupId);
- ListConsumerGroupOffsetsHandler handler = new ListConsumerGroupOffsetsHandler(groupId, options.topicPartitions(), options.requireStable(), logContext);
+ ListConsumerGroupOffsetsHandler.newFuture(groupSpecs.keySet());
+ ListConsumerGroupOffsetsHandler handler =
+ new ListConsumerGroupOffsetsHandler(groupSpecs, options.requireStable(), logContext);
invokeDriver(handler, future, options.timeoutMs);
- return new ListConsumerGroupOffsetsResult(future.get(CoordinatorKey.byGroupId(groupId)));
+ return new ListConsumerGroupOffsetsResult(future.all());
}
@Override
@@ -3756,7 +3758,7 @@ private List getMembersFromGroup(String groupId, String reason)
public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(String groupId,
RemoveMembersFromConsumerGroupOptions options) {
String reason = options.reason() == null || options.reason().isEmpty() ?
- DEFAULT_LEAVE_GROUP_REASON : options.reason();
+ DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
List members;
if (options.removeAll()) {
@@ -4355,16 +4357,27 @@ private QuorumInfo.ReplicaState translateReplicaState(DescribeQuorumResponseData
}
private QuorumInfo createQuorumResult(final DescribeQuorumResponseData.PartitionData partition) {
+ List voters = partition.currentVoters().stream()
+ .map(this::translateReplicaState)
+ .collect(Collectors.toList());
+
+ List observers = partition.observers().stream()
+ .map(this::translateReplicaState)
+ .collect(Collectors.toList());
+
return new QuorumInfo(
- partition.leaderId(),
- partition.currentVoters().stream().map(v -> translateReplicaState(v)).collect(Collectors.toList()),
- partition.observers().stream().map(o -> translateReplicaState(o)).collect(Collectors.toList()));
+ partition.leaderId(),
+ partition.leaderEpoch(),
+ partition.highWatermark(),
+ voters,
+ observers
+ );
}
@Override
DescribeQuorumRequest.Builder createRequest(int timeoutMs) {
return new Builder(DescribeQuorumRequest.singletonRequest(
- new TopicPartition(METADATA_TOPIC_NAME, METADATA_TOPIC_PARTITION.partition())));
+ new TopicPartition(CLUSTER_METADATA_TOPIC_NAME, CLUSTER_METADATA_TOPIC_PARTITION.partition())));
}
@Override
@@ -4380,9 +4393,9 @@ void handleResponse(AbstractResponse response) {
throw new UnknownServerException(msg);
}
DescribeQuorumResponseData.TopicData topic = quorumResponse.data().topics().get(0);
- if (!topic.topicName().equals(METADATA_TOPIC_NAME)) {
+ if (!topic.topicName().equals(CLUSTER_METADATA_TOPIC_NAME)) {
String msg = String.format("DescribeMetadataQuorum received a topic with name %s when %s was expected",
- topic.topicName(), METADATA_TOPIC_NAME);
+ topic.topicName(), CLUSTER_METADATA_TOPIC_NAME);
log.debug(msg);
throw new UnknownServerException(msg);
}
@@ -4393,9 +4406,9 @@ void handleResponse(AbstractResponse response) {
throw new UnknownServerException(msg);
}
DescribeQuorumResponseData.PartitionData partition = topic.partitions().get(0);
- if (partition.partitionIndex() != METADATA_TOPIC_PARTITION.partition()) {
+ if (partition.partitionIndex() != CLUSTER_METADATA_TOPIC_PARTITION.partition()) {
String msg = String.format("DescribeMetadataQuorum received a single partition with index %d when %d was expected",
- partition.partitionIndex(), METADATA_TOPIC_PARTITION.partition());
+ partition.partitionIndex(), CLUSTER_METADATA_TOPIC_PARTITION.partition());
log.debug(msg);
throw new UnknownServerException(msg);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsOptions.java
index 292a47ef393c3..44d3a407327e1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsOptions.java
@@ -23,23 +23,28 @@
import java.util.List;
/**
- * Options for {@link Admin#listConsumerGroupOffsets(String)}.
+ * Options for {@link Admin#listConsumerGroupOffsets(java.util.Map)} and {@link Admin#listConsumerGroupOffsets(String)}.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
public class ListConsumerGroupOffsetsOptions extends AbstractOptions {
- private List topicPartitions = null;
+ private List topicPartitions;
private boolean requireStable = false;
/**
* Set the topic partitions to list as part of the result.
* {@code null} includes all topic partitions.
+ *
+ * @deprecated Since 3.3.
+ * Use {@link Admin#listConsumerGroupOffsets(java.util.Map, ListConsumerGroupOffsetsOptions)}
+ * to specify topic partitions.
*
* @param topicPartitions List of topic partitions to include
* @return This ListGroupOffsetsOptions
*/
+ @Deprecated
public ListConsumerGroupOffsetsOptions topicPartitions(List topicPartitions) {
this.topicPartitions = topicPartitions;
return this;
@@ -55,7 +60,12 @@ public ListConsumerGroupOffsetsOptions requireStable(final boolean requireStable
/**
* Returns a list of topic partitions to add as part of the result.
+ *
+ * @deprecated Since 3.3.
+ * Use {@link Admin#listConsumerGroupOffsets(java.util.Map, ListConsumerGroupOffsetsOptions)}
+ * to specify topic partitions.
*/
+ @Deprecated
public List topicPartitions() {
return topicPartitions;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java
index 48f4531418110..2136e33a401e1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsResult.java
@@ -17,25 +17,32 @@
package org.apache.kafka.clients.admin;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.clients.admin.internals.CoordinatorKey;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.annotation.InterfaceStability;
-import java.util.Map;
-
/**
- * The result of the {@link Admin#listConsumerGroupOffsets(String)} call.
+ * The result of the {@link Admin#listConsumerGroupOffsets(Map)} and
+ * {@link Admin#listConsumerGroupOffsets(String)} call.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Evolving
public class ListConsumerGroupOffsetsResult {
- final KafkaFuture> future;
+ final Map>> futures;
- ListConsumerGroupOffsetsResult(KafkaFuture> future) {
- this.future = future;
+ ListConsumerGroupOffsetsResult(final Map>> futures) {
+ this.futures = futures.entrySet().stream()
+ .collect(Collectors.toMap(e -> e.getKey().idValue, Entry::getValue));
}
/**
@@ -43,7 +50,42 @@ public class ListConsumerGroupOffsetsResult {
* If the group does not have a committed offset for this partition, the corresponding value in the returned map will be null.
*/
public KafkaFuture> partitionsToOffsetAndMetadata() {
- return future;
+ if (futures.size() != 1) {
+ throw new IllegalStateException("Offsets from multiple consumer groups were requested. " +
+ "Use partitionsToOffsetAndMetadata(groupId) instead to get future for a specific group.");
+ }
+ return futures.values().iterator().next();
}
+ /**
+ * Return a future which yields a map of topic partitions to OffsetAndMetadata objects for
+ * the specified group. If the group doesn't have a committed offset for a specific
+ * partition, the corresponding value in the returned map will be null.
+ */
+ public KafkaFuture> partitionsToOffsetAndMetadata(String groupId) {
+ if (!futures.containsKey(groupId))
+ throw new IllegalArgumentException("Offsets for consumer group '" + groupId + "' were not requested.");
+ return futures.get(groupId);
+ }
+
+ /**
+ * Return a future which yields all Map objects,
+ * if requests for all the groups succeed.
+ */
+ public KafkaFuture>> all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).thenApply(
+ nil -> {
+ Map> listedConsumerGroupOffsets = new HashMap<>(futures.size());
+ futures.forEach((key, future) -> {
+ try {
+ listedConsumerGroupOffsets.put(key, future.get());
+ } catch (InterruptedException | ExecutionException e) {
+ // This should be unreachable, since the KafkaFuture#allOf already ensured
+ // that all of the futures completed successfully.
+ throw new RuntimeException(e);
+ }
+ });
+ return listedConsumerGroupOffsets;
+ });
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsSpec.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsSpec.java
new file mode 100644
index 0000000000000..83858e49c8170
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupOffsetsSpec.java
@@ -0,0 +1,79 @@
+/*
+ * 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.admin;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+import java.util.Objects;
+
+/**
+ * Specification of consumer group offsets to list using {@link Admin#listConsumerGroupOffsets(java.util.Map)}.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListConsumerGroupOffsetsSpec {
+
+ private Collection topicPartitions;
+
+ /**
+ * Set the topic partitions whose offsets are to be listed for a consumer group.
+ * {@code null} includes all topic partitions.
+ *
+ * @param topicPartitions List of topic partitions to include
+ * @return This ListConsumerGroupOffsetSpec
+ */
+ public ListConsumerGroupOffsetsSpec topicPartitions(Collection topicPartitions) {
+ this.topicPartitions = topicPartitions;
+ return this;
+ }
+
+ /**
+ * Returns the topic partitions whose offsets are to be listed for a consumer group.
+ * {@code null} indicates that offsets of all partitions of the group are to be listed.
+ */
+ public Collection topicPartitions() {
+ return topicPartitions;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ListConsumerGroupOffsetsSpec)) {
+ return false;
+ }
+ ListConsumerGroupOffsetsSpec that = (ListConsumerGroupOffsetsSpec) o;
+ return Objects.equals(topicPartitions, that.topicPartitions);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(topicPartitions);
+ }
+
+ @Override
+ public String toString() {
+ return "ListConsumerGroupOffsetsSpec(" +
+ "topicPartitions=" + topicPartitions +
+ ')';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java b/clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java
index 75476d77dcff1..f9e4f8c11c904 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java
@@ -24,20 +24,38 @@
* This class is used to describe the state of the quorum received in DescribeQuorumResponse.
*/
public class QuorumInfo {
- private final Integer leaderId;
+ private final int leaderId;
+ private final long leaderEpoch;
+ private final long highWatermark;
private final List voters;
private final List observers;
- QuorumInfo(Integer leaderId, List voters, List observers) {
+ QuorumInfo(
+ int leaderId,
+ long leaderEpoch,
+ long highWatermark,
+ List voters,
+ List observers
+ ) {
this.leaderId = leaderId;
+ this.leaderEpoch = leaderEpoch;
+ this.highWatermark = highWatermark;
this.voters = voters;
this.observers = observers;
}
- public Integer leaderId() {
+ public int leaderId() {
return leaderId;
}
+ public long leaderEpoch() {
+ return leaderEpoch;
+ }
+
+ public long highWatermark() {
+ return highWatermark;
+ }
+
public List voters() {
return voters;
}
@@ -51,20 +69,24 @@ public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
QuorumInfo that = (QuorumInfo) o;
- return leaderId.equals(that.leaderId)
- && voters.equals(that.voters)
- && observers.equals(that.observers);
+ return leaderId == that.leaderId
+ && leaderEpoch == that.leaderEpoch
+ && highWatermark == that.highWatermark
+ && Objects.equals(voters, that.voters)
+ && Objects.equals(observers, that.observers);
}
@Override
public int hashCode() {
- return Objects.hash(leaderId, voters, observers);
+ return Objects.hash(leaderId, leaderEpoch, highWatermark, voters, observers);
}
@Override
public String toString() {
return "QuorumInfo(" +
"leaderId=" + leaderId +
+ ", leaderEpoch=" + leaderEpoch +
+ ", highWatermark=" + highWatermark +
", voters=" + voters +
", observers=" + observers +
')';
@@ -73,8 +95,8 @@ public String toString() {
public static class ReplicaState {
private final int replicaId;
private final long logEndOffset;
- private final OptionalLong lastFetchTimeMs;
- private final OptionalLong lastCaughtUpTimeMs;
+ private final OptionalLong lastFetchTimestamp;
+ private final OptionalLong lastCaughtUpTimestamp;
ReplicaState() {
this(0, 0, OptionalLong.empty(), OptionalLong.empty());
@@ -83,13 +105,13 @@ public static class ReplicaState {
ReplicaState(
int replicaId,
long logEndOffset,
- OptionalLong lastFetchTimeMs,
- OptionalLong lastCaughtUpTimeMs
+ OptionalLong lastFetchTimestamp,
+ OptionalLong lastCaughtUpTimestamp
) {
this.replicaId = replicaId;
this.logEndOffset = logEndOffset;
- this.lastFetchTimeMs = lastFetchTimeMs;
- this.lastCaughtUpTimeMs = lastCaughtUpTimeMs;
+ this.lastFetchTimestamp = lastFetchTimestamp;
+ this.lastCaughtUpTimestamp = lastCaughtUpTimestamp;
}
/**
@@ -109,19 +131,21 @@ public long logEndOffset() {
}
/**
- * Return the lastFetchTime in milliseconds for this replica.
+ * Return the last millisecond timestamp that the leader received a
+ * fetch from this replica.
* @return The value of the lastFetchTime if known, empty otherwise
*/
- public OptionalLong lastFetchTimeMs() {
- return lastFetchTimeMs;
+ public OptionalLong lastFetchTimestamp() {
+ return lastFetchTimestamp;
}
/**
- * Return the lastCaughtUpTime in milliseconds for this replica.
+ * Return the last millisecond timestamp at which this replica was known to be
+ * caught up with the leader.
* @return The value of the lastCaughtUpTime if known, empty otherwise
*/
- public OptionalLong lastCaughtUpTimeMs() {
- return lastCaughtUpTimeMs;
+ public OptionalLong lastCaughtUpTimestamp() {
+ return lastCaughtUpTimestamp;
}
@Override
@@ -131,13 +155,13 @@ public boolean equals(Object o) {
ReplicaState that = (ReplicaState) o;
return replicaId == that.replicaId
&& logEndOffset == that.logEndOffset
- && lastFetchTimeMs.equals(that.lastFetchTimeMs)
- && lastCaughtUpTimeMs.equals(that.lastCaughtUpTimeMs);
+ && lastFetchTimestamp.equals(that.lastFetchTimestamp)
+ && lastCaughtUpTimestamp.equals(that.lastCaughtUpTimestamp);
}
@Override
public int hashCode() {
- return Objects.hash(replicaId, logEndOffset, lastFetchTimeMs, lastCaughtUpTimeMs);
+ return Objects.hash(replicaId, logEndOffset, lastFetchTimestamp, lastCaughtUpTimestamp);
}
@Override
@@ -145,8 +169,8 @@ public String toString() {
return "ReplicaState(" +
"replicaId=" + replicaId +
", logEndOffset=" + logEndOffset +
- ", lastFetchTimeMs=" + lastFetchTimeMs +
- ", lastCaughtUpTimeMs=" + lastCaughtUpTimeMs +
+ ", lastFetchTimestamp=" + lastFetchTimestamp +
+ ", lastCaughtUpTimestamp=" + lastCaughtUpTimestamp +
')';
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/SupportedVersionRange.java b/clients/src/main/java/org/apache/kafka/clients/admin/SupportedVersionRange.java
index d71da31fb8200..b85a392a658c3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/SupportedVersionRange.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/SupportedVersionRange.java
@@ -36,10 +36,10 @@ public class SupportedVersionRange {
* @throws IllegalArgumentException Raised when the condition described above is not met.
*/
SupportedVersionRange(final short minVersion, final short maxVersion) {
- if (minVersion < 1 || maxVersion < 1 || maxVersion < minVersion) {
+ if (minVersion < 0 || maxVersion < 0 || maxVersion < minVersion) {
throw new IllegalArgumentException(
String.format(
- "Expected 1 <= minVersion <= maxVersion but received minVersion:%d, maxVersion:%d.",
+ "Expected 0 <= minVersion <= maxVersion but received minVersion:%d, maxVersion:%d.",
minVersion,
maxVersion));
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
index d00db4b18c694..0e1b03d964fea 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
@@ -21,6 +21,7 @@
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.FindCoordinatorRequest.NoBatchedFindCoordinatorsException;
+import org.apache.kafka.common.requests.OffsetFetchRequest.NoBatchedOffsetFetchRequestException;
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
@@ -253,7 +254,7 @@ public void onFailure(
.collect(Collectors.toSet());
retryLookup(keysToUnmap);
- } else if (t instanceof NoBatchedFindCoordinatorsException) {
+ } else if (t instanceof NoBatchedFindCoordinatorsException || t instanceof NoBatchedOffsetFetchRequestException) {
((CoordinatorStrategy) handler.lookupStrategy()).disableBatch();
Set keysToUnmap = spec.keys.stream()
.filter(future.lookupKeys()::contains)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
index e6fc0d624a0cd..02b68527c3c7e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/CoordinatorStrategy.java
@@ -120,6 +120,10 @@ public void disableBatch() {
batch = false;
}
+ public boolean batch() {
+ return batch;
+ }
+
private CoordinatorKey requireSingletonAndType(Set keys) {
if (keys.size() != 1) {
throw new IllegalArgumentException("Unexpected size of key set: expected 1, but got " + keys.size());
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
index 08648821f7c9f..21c7d8d488f3d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
@@ -17,14 +17,16 @@
package org.apache.kafka.clients.admin.internals;
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.stream.Collectors;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
@@ -36,39 +38,26 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-public class ListConsumerGroupOffsetsHandler extends AdminApiHandler.Batched> {
+public class ListConsumerGroupOffsetsHandler implements AdminApiHandler> {
- private final CoordinatorKey groupId;
- private final List partitions;
private final boolean requireStable;
+ private final Map groupSpecs;
private final Logger log;
- private final AdminApiLookupStrategy lookupStrategy;
+ private final CoordinatorStrategy lookupStrategy;
public ListConsumerGroupOffsetsHandler(
- String groupId,
- List partitions,
- LogContext logContext
- ) {
- this(groupId, partitions, false, logContext);
- }
-
- public ListConsumerGroupOffsetsHandler(
- String groupId,
- List partitions,
+ Map groupSpecs,
boolean requireStable,
LogContext logContext
) {
- this.groupId = CoordinatorKey.byGroupId(groupId);
- this.partitions = partitions;
- this.requireStable = requireStable;
this.log = logContext.logger(ListConsumerGroupOffsetsHandler.class);
this.lookupStrategy = new CoordinatorStrategy(CoordinatorType.GROUP, logContext);
+ this.groupSpecs = groupSpecs;
+ this.requireStable = requireStable;
}
- public static AdminApiFuture.SimpleAdminApiFuture> newFuture(
- String groupId
- ) {
- return AdminApiFuture.forKeys(Collections.singleton(CoordinatorKey.byGroupId(groupId)));
+ public static AdminApiFuture.SimpleAdminApiFuture> newFuture(Collection groupIds) {
+ return AdminApiFuture.forKeys(coordinatorKeys(groupIds));
}
@Override
@@ -82,16 +71,45 @@ public AdminApiLookupStrategy lookupStrategy() {
}
private void validateKeys(Set groupIds) {
- if (!groupIds.equals(Collections.singleton(groupId))) {
+ Set keys = coordinatorKeys(groupSpecs.keySet());
+ if (!keys.containsAll(groupIds)) {
throw new IllegalArgumentException("Received unexpected group ids " + groupIds +
- " (expected only " + Collections.singleton(groupId) + ")");
+ " (expected one of " + keys + ")");
}
}
+ private static Set coordinatorKeys(Collection groupIds) {
+ return groupIds.stream()
+ .map(CoordinatorKey::byGroupId)
+ .collect(Collectors.toSet());
+ }
+
+ public OffsetFetchRequest.Builder buildBatchedRequest(Set groupIds) {
+ // Create a map that only contains the consumer groups owned by the coordinator.
+ Map> coordinatorGroupIdToTopicPartitions = new HashMap<>(groupIds.size());
+ groupIds.forEach(g -> {
+ ListConsumerGroupOffsetsSpec spec = groupSpecs.get(g.idValue);
+ List partitions = spec.topicPartitions() != null ? new ArrayList<>(spec.topicPartitions()) : null;
+ coordinatorGroupIdToTopicPartitions.put(g.idValue, partitions);
+ });
+
+ return new OffsetFetchRequest.Builder(coordinatorGroupIdToTopicPartitions, requireStable, false);
+ }
+
@Override
- public OffsetFetchRequest.Builder buildBatchedRequest(int coordinatorId, Set groupIds) {
+ public Collection> buildRequest(int brokerId, Set groupIds) {
validateKeys(groupIds);
- return new OffsetFetchRequest.Builder(groupId.idValue, requireStable, partitions, false);
+
+ // When the OffsetFetchRequest fails with NoBatchedOffsetFetchRequestException, we completely disable
+ // the batching end-to-end, including the FindCoordinatorRequest.
+ if (lookupStrategy.batch()) {
+ return Collections.singletonList(new RequestAndKeys<>(buildBatchedRequest(groupIds), groupIds));
+ } else {
+ return groupIds.stream().map(groupId -> {
+ Set keys = Collections.singleton(groupId);
+ return new RequestAndKeys<>(buildBatchedRequest(keys), keys);
+ }).collect(Collectors.toList());
+ }
}
@Override
@@ -104,44 +122,46 @@ public ApiResult> handleR
final OffsetFetchResponse response = (OffsetFetchResponse) abstractResponse;
- // the groupError will contain the group level error for v0-v8 OffsetFetchResponse
- Errors groupError = response.groupLevelError(groupId.idValue);
- if (groupError != Errors.NONE) {
- final Map failed = new HashMap<>();
- final Set groupsToUnmap = new HashSet<>();
-
- handleGroupError(groupId, groupError, failed, groupsToUnmap);
-
- return new ApiResult<>(Collections.emptyMap(), failed, new ArrayList<>(groupsToUnmap));
- } else {
- final Map groupOffsetsListing = new HashMap<>();
-
- response.partitionDataMap(groupId.idValue).forEach((topicPartition, partitionData) -> {
- final Errors error = partitionData.error;
- if (error == Errors.NONE) {
- final long offset = partitionData.offset;
- final String metadata = partitionData.metadata;
- final Optional leaderEpoch = partitionData.leaderEpoch;
- // Negative offset indicates that the group has no committed offset for this partition
- if (offset < 0) {
- groupOffsetsListing.put(topicPartition, null);
+ Map> completed = new HashMap<>();
+ Map failed = new HashMap<>();
+ List unmapped = new ArrayList<>();
+ for (CoordinatorKey coordinatorKey : groupIds) {
+ String group = coordinatorKey.idValue;
+ if (response.groupHasError(group)) {
+ handleGroupError(CoordinatorKey.byGroupId(group), response.groupLevelError(group), failed, unmapped);
+ } else {
+ final Map groupOffsetsListing = new HashMap<>();
+ Map responseData = response.partitionDataMap(group);
+ for (Map.Entry partitionEntry : responseData.entrySet()) {
+ final TopicPartition topicPartition = partitionEntry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = partitionEntry.getValue();
+ final Errors error = partitionData.error;
+
+ if (error == Errors.NONE) {
+ final long offset = partitionData.offset;
+ final String metadata = partitionData.metadata;
+ final Optional leaderEpoch = partitionData.leaderEpoch;
+ // Negative offset indicates that the group has no committed offset for this partition
+ if (offset < 0) {
+ groupOffsetsListing.put(topicPartition, null);
+ } else {
+ groupOffsetsListing.put(topicPartition, new OffsetAndMetadata(offset, leaderEpoch, metadata));
+ }
} else {
- groupOffsetsListing.put(topicPartition, new OffsetAndMetadata(offset, leaderEpoch, metadata));
+ log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
}
- } else {
- log.warn("Skipping return offset for {} due to error {}.", topicPartition, error);
}
- });
-
- return ApiResult.completed(groupId, groupOffsetsListing);
+ completed.put(CoordinatorKey.byGroupId(group), groupOffsetsListing);
+ }
}
+ return new ApiResult<>(completed, failed, unmapped);
}
private void handleGroupError(
CoordinatorKey groupId,
Errors error,
Map failed,
- Set groupsToUnmap
+ List groupsToUnmap
) {
switch (error) {
case GROUP_AUTHORIZATION_FAILED:
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index fa80727d5af43..80b515d601b2a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -187,11 +187,12 @@ public AbstractCoordinator(GroupRebalanceConfig rebalanceConfig,
/**
* Invoked prior to each group join or rejoin. This is typically used to perform any
* cleanup from the previous generation (such as committing offsets for the consumer)
+ * @param timer Timer bounding how long this method can block
* @param generation The previous generation or -1 if there was none
* @param memberId The identifier of this member in the previous group or "" if there was none
* @return true If onJoinPrepare async commit succeeded, false otherwise
*/
- protected abstract boolean onJoinPrepare(int generation, String memberId);
+ protected abstract boolean onJoinPrepare(Timer timer, int generation, String memberId);
/**
* Invoked when the leader is elected. This is used by the leader to perform the assignment
@@ -230,14 +231,27 @@ protected abstract void onJoinComplete(int generation,
protected void onLeavePrepare() {}
/**
- * Visible for testing.
- *
* Ensure that the coordinator is ready to receive requests.
*
* @param timer Timer bounding how long this method can block
* @return true If coordinator discovery and initial connection succeeded, false otherwise
*/
protected synchronized boolean ensureCoordinatorReady(final Timer timer) {
+ return ensureCoordinatorReady(timer, false);
+ }
+
+ /**
+ * Ensure that the coordinator is ready to receive requests. This will return
+ * immediately without blocking. It is intended to be called in an asynchronous
+ * context when wakeups are not expected.
+ *
+ * @return true If coordinator discovery and initial connection succeeded, false otherwise
+ */
+ protected synchronized boolean ensureCoordinatorReadyAsync() {
+ return ensureCoordinatorReady(time.timer(0), true);
+ }
+
+ private synchronized boolean ensureCoordinatorReady(final Timer timer, boolean disableWakeup) {
if (!coordinatorUnknown())
return true;
@@ -248,7 +262,7 @@ protected synchronized boolean ensureCoordinatorReady(final Timer timer) {
throw fatalException;
}
final RequestFuture future = lookupCoordinator();
- client.poll(future, timer);
+ client.poll(future, timer, disableWakeup);
if (!future.isDone()) {
// ran out of time
@@ -426,7 +440,7 @@ boolean joinGroupIfNeeded(final Timer timer) {
// exception, in which case upon retry we should not retry onJoinPrepare either.
needsJoinPrepare = false;
// return false when onJoinPrepare is waiting for committing offset
- if (!onJoinPrepare(generation.generationId, generation.memberId)) {
+ if (!onJoinPrepare(timer, generation.generationId, generation.memberId)) {
needsJoinPrepare = true;
//should not initiateJoinGroup if needsJoinPrepare still is true
return false;
@@ -478,11 +492,11 @@ boolean joinGroupIfNeeded(final Timer timer) {
resetJoinGroupFuture();
synchronized (AbstractCoordinator.this) {
- final String shortReason = String.format("rebalance failed due to %s",
- exception.getClass().getSimpleName());
+ final String simpleName = exception.getClass().getSimpleName();
+ final String shortReason = String.format("rebalance failed due to %s", simpleName);
final String fullReason = String.format("rebalance failed due to '%s' (%s)",
exception.getMessage(),
- exception.getClass().getSimpleName());
+ simpleName);
requestRejoin(shortReason, fullReason);
}
@@ -559,7 +573,7 @@ RequestFuture sendJoinGroupRequest() {
.setProtocolType(protocolType())
.setProtocols(metadata())
.setRebalanceTimeoutMs(this.rebalanceConfig.rebalanceTimeoutMs)
- .setReason(this.rejoinReason)
+ .setReason(JoinGroupRequest.maybeTruncateReason(this.rejoinReason))
);
log.debug("Sending JoinGroup ({}) to coordinator {}", requestBuilder, this.coordinator);
@@ -811,9 +825,6 @@ public void handle(SyncGroupResponse syncResponse,
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
- // consumer didn't get assignment in this generation, so we need to reset generation
- // to avoid joinGroup with out-of-data ownedPartitions in cooperative rebalance
- resetStateOnResponseError(ApiKeys.SYNC_GROUP, error, false);
future.raise(error);
} else if (error == Errors.FENCED_INSTANCE_ID) {
// for sync-group request, even if the generation has changed we would not expect the instance id
@@ -1114,7 +1125,7 @@ public synchronized RequestFuture maybeLeaveGroup(String leaveReason) {
generation.memberId, coordinator, leaveReason);
LeaveGroupRequest.Builder request = new LeaveGroupRequest.Builder(
rebalanceConfig.groupId,
- Collections.singletonList(new MemberIdentity().setMemberId(generation.memberId).setReason(leaveReason))
+ Collections.singletonList(new MemberIdentity().setMemberId(generation.memberId).setReason(JoinGroupRequest.maybeTruncateReason(leaveReason)))
);
future = client.send(coordinator, request).compose(new LeaveGroupResponseHandler(generation));
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index b853ff99e8231..5eb7d87819795 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.clients.consumer.internals;
-import java.time.Duration;
import java.util.SortedSet;
import java.util.TreeSet;
import org.apache.kafka.clients.GroupRebalanceConfig;
@@ -141,6 +140,12 @@ private boolean sameRequest(final Set currentRequest, final Gene
}
private final RebalanceProtocol protocol;
+ // pending commit offset request in onJoinPrepare
+ private RequestFuture autoCommitOffsetRequestFuture = null;
+ // a timer for join prepare to know when to stop.
+ // it'll set to rebalance timeout so that the member can join the group successfully
+ // even though offset commit failed.
+ private Timer joinPrepareTimer = null;
/**
* Initialize the coordination manager.
@@ -483,10 +488,14 @@ void maybeUpdateSubscriptionMetadata() {
}
}
- private boolean coordinatorUnknownAndUnready(Timer timer) {
+ private boolean coordinatorUnknownAndUnreadySync(Timer timer) {
return coordinatorUnknown() && !ensureCoordinatorReady(timer);
}
+ private boolean coordinatorUnknownAndUnreadyAsync() {
+ return coordinatorUnknown() && !ensureCoordinatorReadyAsync();
+ }
+
/**
* Poll for coordinator events. This ensures that the coordinator is known and that the consumer
* has joined the group (if it is using group management). This also handles periodic offset commits
@@ -512,7 +521,7 @@ public boolean poll(Timer timer, boolean waitForJoinGroup) {
// Always update the heartbeat last poll time so that the heartbeat thread does not leave the
// group proactively due to application inactivity even if (say) the coordinator cannot be found.
pollHeartbeat(timer.currentTimeMs());
- if (coordinatorUnknownAndUnready(timer)) {
+ if (coordinatorUnknownAndUnreadySync(timer)) {
return false;
}
@@ -740,24 +749,59 @@ private void validateCooperativeAssignment(final Map future = maybeAutoCommitOffsetsAsync();
- // return true when
- // 1. future is null, which means no commit request sent, so it is still considered completed
- // 2. offset commit completed
- // 3. offset commit failed with non-retriable exception
- if (future == null)
- onJoinPrepareAsyncCommitCompleted = true;
- else if (future.succeeded())
- onJoinPrepareAsyncCommitCompleted = true;
- else if (future.failed() && !future.isRetriable()) {
- log.error("Asynchronous auto-commit of offsets failed: {}", future.exception().getMessage());
- onJoinPrepareAsyncCommitCompleted = true;
+ // and there is no in-flight offset commit request
+ if (autoCommitEnabled && autoCommitOffsetRequestFuture == null) {
+ maybeMarkPartitionsPendingRevocation();
+ autoCommitOffsetRequestFuture = maybeAutoCommitOffsetsAsync();
}
+ // wait for commit offset response before timer expired
+ if (autoCommitOffsetRequestFuture != null) {
+ Timer pollTimer = timer.remainingMs() < joinPrepareTimer.remainingMs() ?
+ timer : joinPrepareTimer;
+ client.poll(autoCommitOffsetRequestFuture, pollTimer);
+ joinPrepareTimer.update();
+
+ // Keep retrying/waiting the offset commit when:
+ // 1. offset commit haven't done (and joinPrepareTimer not expired)
+ // 2. failed with retryable exception (and joinPrepareTimer not expired)
+ // Otherwise, continue to revoke partitions, ex:
+ // 1. if joinPrepareTime has expired
+ // 2. if offset commit failed with no-retryable exception
+ // 3. if offset commit success
+ boolean onJoinPrepareAsyncCommitCompleted = true;
+ if (joinPrepareTimer.isExpired()) {
+ log.error("Asynchronous auto-commit of offsets failed: joinPrepare timeout. Will continue to join group");
+ } else if (!autoCommitOffsetRequestFuture.isDone()) {
+ onJoinPrepareAsyncCommitCompleted = false;
+ } else if (autoCommitOffsetRequestFuture.failed() && autoCommitOffsetRequestFuture.isRetriable()) {
+ log.debug("Asynchronous auto-commit of offsets failed with retryable error: {}. Will retry it.",
+ autoCommitOffsetRequestFuture.exception().getMessage());
+ onJoinPrepareAsyncCommitCompleted = false;
+ } else if (autoCommitOffsetRequestFuture.failed() && !autoCommitOffsetRequestFuture.isRetriable()) {
+ log.error("Asynchronous auto-commit of offsets failed: {}. Will continue to join group.",
+ autoCommitOffsetRequestFuture.exception().getMessage());
+ }
+ if (autoCommitOffsetRequestFuture.isDone()) {
+ autoCommitOffsetRequestFuture = null;
+ }
+ if (!onJoinPrepareAsyncCommitCompleted) {
+ pollTimer.sleep(Math.min(pollTimer.remainingMs(), rebalanceConfig.retryBackoffMs));
+ timer.update();
+ return false;
+ }
+ }
// the generation / member-id can possibly be reset by the heartbeat thread
// upon getting errors or heartbeat timeouts; in this case whatever is previously
@@ -809,11 +853,30 @@ else if (future.failed() && !future.isRetriable()) {
isLeader = false;
subscriptions.resetGroupSubscription();
+ joinPrepareTimer = null;
+ autoCommitOffsetRequestFuture = null;
+ timer.update();
if (exception != null) {
throw new KafkaException("User rebalance callback throws an error", exception);
}
- return onJoinPrepareAsyncCommitCompleted;
+ return true;
+ }
+
+ private void maybeMarkPartitionsPendingRevocation() {
+ if (protocol != RebalanceProtocol.EAGER) {
+ return;
+ }
+
+ // When asynchronously committing offsets prior to the revocation of a set of partitions, there will be a
+ // window of time between when the offset commit is sent and when it returns and revocation completes. It is
+ // possible for pending fetches for these partitions to return during this time, which means the application's
+ // position may get ahead of the committed position prior to revocation. This can cause duplicate consumption.
+ // To prevent this, we mark the partitions as "pending revocation," which stops the Fetcher from sending new
+ // fetches or returning data from previous fetches to the user.
+ Set partitions = subscriptions.assignedPartitions();
+ log.debug("Marking assigned partitions pending for revocation: {}", partitions);
+ subscriptions.markPendingRevocation(partitions);
}
@Override
@@ -1009,7 +1072,7 @@ public RequestFuture commitOffsetsAsync(final Map offsets,
return true;
do {
- if (coordinatorUnknownAndUnready(timer)) {
+ if (coordinatorUnknownAndUnreadySync(timer)) {
return false;
}
@@ -1229,8 +1292,10 @@ RequestFuture sendOffsetCommitRequest(final Map sendOffsetCommitRequest(final Map sendOffsetCommitRequest(final Map(requestTopicDataMap.values()))
);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
index 4b9112016e986..6646dc6c893e0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
@@ -211,8 +211,23 @@ public void poll(RequestFuture future) {
* @throws InterruptException if the calling thread is interrupted
*/
public boolean poll(RequestFuture future, Timer timer) {
+ return poll(future, timer, false);
+ }
+
+ /**
+ * Block until the provided request future request has finished or the timeout has expired.
+ *
+ * @param future The request future to wait for
+ * @param timer Timer bounding how long this method can block
+ * @param disableWakeup true if we should not check for wakeups, false otherwise
+ *
+ * @return true if the future is done, false otherwise
+ * @throws WakeupException if {@link #wakeup()} is called from another thread and `disableWakeup` is false
+ * @throws InterruptException if the calling thread is interrupted
+ */
+ public boolean poll(RequestFuture future, Timer timer, boolean disableWakeup) {
do {
- poll(timer, future);
+ poll(timer, future, disableWakeup);
} while (!future.isDone() && timer.notExpired());
return future.isDone();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index 73ffd217efe21..c93b675f755f2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -346,6 +346,7 @@ public void onFailure(RuntimeException e) {
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler != null) {
handler.handleError(e);
+ handler.sessionTopicPartitions().forEach(subscriptions::clearPreferredReadReplica);
}
} finally {
nodesWithPendingFetchRequests.remove(fetchTarget.id());
@@ -1154,7 +1155,9 @@ Node selectReadReplica(TopicPartition partition, Node leaderReplica, long curren
} else {
log.trace("Not fetching from {} for partition {} since it is marked offline or is missing from our metadata," +
" using the leader instead.", nodeId, partition);
- subscriptions.clearPreferredReadReplica(partition);
+ // Note that this condition may happen due to stale metadata, so we clear preferred replica and
+ // refresh metadata.
+ requestMetadataUpdate(partition);
return leaderReplica;
}
} else {
@@ -1335,16 +1338,16 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
error == Errors.FENCED_LEADER_EPOCH ||
error == Errors.OFFSET_NOT_AVAILABLE) {
log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
- this.metadata.requestUpdate();
+ requestMetadataUpdate(tp);
} else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
log.warn("Received unknown topic or partition error in fetch for partition {}", tp);
- this.metadata.requestUpdate();
+ requestMetadataUpdate(tp);
} else if (error == Errors.UNKNOWN_TOPIC_ID) {
log.warn("Received unknown topic ID error in fetch for partition {}", tp);
- this.metadata.requestUpdate();
+ requestMetadataUpdate(tp);
} else if (error == Errors.INCONSISTENT_TOPIC_ID) {
log.warn("Received inconsistent topic ID error in fetch for partition {}", tp);
- this.metadata.requestUpdate();
+ requestMetadataUpdate(tp);
} else if (error == Errors.OFFSET_OUT_OF_RANGE) {
Optional clearedReplicaId = subscriptions.clearPreferredReadReplica(tp);
if (!clearedReplicaId.isPresent()) {
@@ -1944,4 +1947,9 @@ private Set topicsForPartitions(Collection partitions) {
return partitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
}
+ private void requestMetadataUpdate(TopicPartition topicPartition) {
+ this.metadata.requestUpdate();
+ this.subscriptions.clearPreferredReadReplica(topicPartition);
+ }
+
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
index 0225822a19466..b41122f0e5060 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
@@ -620,7 +620,12 @@ public synchronized Optional preferredReadReplica(TopicPartition tp, lo
* @return the removed preferred read replica if set, None otherwise.
*/
public synchronized Optional clearPreferredReadReplica(TopicPartition tp) {
- return assignedState(tp).clearPreferredReadReplica();
+ final TopicPartitionState topicPartitionState = assignedStateOrNull(tp);
+ if (topicPartitionState == null) {
+ return Optional.empty();
+ } else {
+ return topicPartitionState.clearPreferredReadReplica();
+ }
}
public synchronized Map allConsumed() {
@@ -738,6 +743,10 @@ public synchronized void pause(TopicPartition tp) {
assignedState(tp).pause();
}
+ public synchronized void markPendingRevocation(Set tps) {
+ tps.forEach(tp -> assignedState(tp).markPendingRevocation());
+ }
+
public synchronized void resume(TopicPartition tp) {
assignedState(tp).resume();
}
@@ -769,6 +778,7 @@ private static class TopicPartitionState {
private Long logStartOffset; // the log start offset
private Long lastStableOffset;
private boolean paused; // whether this partition has been paused by the user
+ private boolean pendingRevocation;
private OffsetResetStrategy resetStrategy; // the strategy to use if the offset needs resetting
private Long nextRetryTimeMs;
private Integer preferredReadReplica;
@@ -777,6 +787,7 @@ private static class TopicPartitionState {
TopicPartitionState() {
this.paused = false;
+ this.pendingRevocation = false;
this.endOffsetRequested = false;
this.fetchState = FetchStates.INITIALIZING;
this.position = null;
@@ -966,12 +977,16 @@ private void pause() {
this.paused = true;
}
+ private void markPendingRevocation() {
+ this.pendingRevocation = true;
+ }
+
private void resume() {
this.paused = false;
}
private boolean isFetchable() {
- return !paused && hasValidPosition();
+ return !paused && !pendingRevocation && hasValidPosition();
}
private void highWatermark(Long highWatermark) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 74d408d9a5a49..148023e3a56ef 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -425,8 +425,11 @@ private void warnIfPartitionerDeprecated() {
enableAdaptivePartitioning,
config.getLong(ProducerConfig.PARTITIONER_AVAILABILITY_TIMEOUT_MS_CONFIG)
);
+ // As per Kafka producer configuration documentation batch.size may be set to 0 to explicitly disable
+ // batching which in practice actually means using a batch size of 1.
+ int batchSize = Math.max(1, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
this.accumulator = new RecordAccumulator(logContext,
- config.getInt(ProducerConfig.BATCH_SIZE_CONFIG),
+ batchSize,
this.compressionType,
lingerMs(config),
retryBackoffMs,
@@ -437,7 +440,7 @@ private void warnIfPartitionerDeprecated() {
time,
apiVersions,
transactionManager,
- new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME));
+ new BufferPool(this.totalMemorySize, batchSize, metrics, time, PRODUCER_METRIC_GROUP_NAME));
List addresses = ClientUtils.parseAndValidateAddresses(
config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
@@ -1465,13 +1468,21 @@ public boolean isDone() {
private class AppendCallbacks implements RecordAccumulator.AppendCallbacks {
private final Callback userCallback;
private final ProducerInterceptors interceptors;
- private final ProducerRecord record;
- protected int partition = RecordMetadata.UNKNOWN_PARTITION;
+ private final String topic;
+ private final Integer recordPartition;
+ private final String recordLogString;
+ private volatile int partition = RecordMetadata.UNKNOWN_PARTITION;
+ private volatile TopicPartition topicPartition;
private AppendCallbacks(Callback userCallback, ProducerInterceptors interceptors, ProducerRecord record) {
this.userCallback = userCallback;
this.interceptors = interceptors;
- this.record = record;
+ // Extract record info as we don't want to keep a reference to the record during
+ // whole lifetime of the batch.
+ // We don't want to have an NPE here, because the interceptors would not be notified (see .doSend).
+ topic = record != null ? record.topic() : null;
+ recordPartition = record != null ? record.partition() : null;
+ recordLogString = log.isTraceEnabled() && record != null ? record.toString() : "";
}
@Override
@@ -1491,7 +1502,7 @@ public void setPartition(int partition) {
if (log.isTraceEnabled()) {
// Log the message here, because we don't know the partition before that.
- log.trace("Attempting to append record {} with callback {} to topic {} partition {}", record, userCallback, record.topic(), partition);
+ log.trace("Attempting to append record {} with callback {} to topic {} partition {}", recordLogString, userCallback, topic, partition);
}
}
@@ -1500,11 +1511,15 @@ public int getPartition() {
}
public TopicPartition topicPartition() {
- if (record == null)
- return null;
- return partition == RecordMetadata.UNKNOWN_PARTITION
- ? ProducerInterceptors.extractTopicPartition(record)
- : new TopicPartition(record.topic(), partition);
+ if (topicPartition == null && topic != null) {
+ if (partition != RecordMetadata.UNKNOWN_PARTITION)
+ topicPartition = new TopicPartition(topic, partition);
+ else if (recordPartition != null)
+ topicPartition = new TopicPartition(topic, recordPartition);
+ else
+ topicPartition = new TopicPartition(topic, RecordMetadata.UNKNOWN_PARTITION);
+ }
+ return topicPartition;
}
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
index aff5e49cfcb1b..a3ae15a3baf0d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
@@ -105,7 +105,7 @@ public class ProducerConfig extends AbstractConfig {
"If a broker cannot process produce requests from a partition for " + PARTITIONER_AVAILABILITY_TIMEOUT_MS_CONFIG + " time, "
+ "the partitioner treats that partition as not available. If the value is 0, this logic is disabled. "
+ "Note: this setting has no effect if a custom partitioner is used or " + PARTITIONER_ADPATIVE_PARTITIONING_ENABLE_CONFIG
- + " is set to 'false'";
/** partitioner.ignore.keys */
public static final String PARTITIONER_IGNORE_KEYS_CONFIG = "partitioner.ignore.keys";
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
index a5805df56b78d..a1493d2edd66a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
@@ -56,6 +56,9 @@ public class BuiltInPartitioner {
public BuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
this.log = logContext.logger(BuiltInPartitioner.class);
this.topic = topic;
+ if (stickyBatchSize < 1) {
+ throw new IllegalArgumentException("stickyBatchSize must be >= 1 but got " + stickyBatchSize);
+ }
this.stickyBatchSize = stickyBatchSize;
}
@@ -170,13 +173,52 @@ boolean isPartitionChanged(StickyPartitionInfo partitionInfo) {
* @param cluster The cluster information
*/
void updatePartitionInfo(StickyPartitionInfo partitionInfo, int appendedBytes, Cluster cluster) {
+ updatePartitionInfo(partitionInfo, appendedBytes, cluster, true);
+ }
+
+ /**
+ * Update partition info with the number of bytes appended and maybe switch partition.
+ * NOTE this function needs to be called under the partition's batch queue lock.
+ *
+ * @param partitionInfo The sticky partition info object returned by peekCurrentPartitionInfo
+ * @param appendedBytes The number of bytes appended to this partition
+ * @param cluster The cluster information
+ * @param enableSwitch If true, switch partition once produced enough bytes
+ */
+ void updatePartitionInfo(StickyPartitionInfo partitionInfo, int appendedBytes, Cluster cluster, boolean enableSwitch) {
// partitionInfo may be null if the caller didn't use built-in partitioner.
if (partitionInfo == null)
return;
assert partitionInfo == stickyPartitionInfo.get();
int producedBytes = partitionInfo.producedBytes.addAndGet(appendedBytes);
- if (producedBytes >= stickyBatchSize) {
+
+ // We're trying to switch partition once we produce stickyBatchSize bytes to a partition
+ // but doing so may hinder batching because partition switch may happen while batch isn't
+ // ready to send. This situation is especially likely with high linger.ms setting.
+ // Consider the following example:
+ // linger.ms=500, producer produces 12KB in 500ms, batch.size=16KB
+ // - first batch collects 12KB in 500ms, gets sent
+ // - second batch collects 4KB, then we switch partition, so 4KB gets eventually sent
+ // - ... and so on - we'd get 12KB and 4KB batches
+ // To get more optimal batching and avoid 4KB fractional batches, the caller may disallow
+ // partition switch if batch is not ready to send, so with the example above we'd avoid
+ // fractional 4KB batches: in that case the scenario would look like this:
+ // - first batch collects 12KB in 500ms, gets sent
+ // - second batch collects 4KB, but partition switch doesn't happen because batch in not ready
+ // - second batch collects 12KB in 500ms, gets sent and now we switch partition.
+ // - ... and so on - we'd just send 12KB batches
+ // We cap the produced bytes to not exceed 2x of the batch size to avoid pathological cases
+ // (e.g. if we have a mix of keyed and unkeyed messages, key messages may create an
+ // unready batch after the batch that disabled partition switch becomes ready).
+ // As a result, with high latency.ms setting we end up switching partitions after producing
+ // between stickyBatchSize and stickyBatchSize * 2 bytes, to better align with batch boundary.
+ if (producedBytes >= stickyBatchSize * 2) {
+ log.trace("Produced {} bytes, exceeding twice the batch size of {} bytes, with switching set to {}",
+ producedBytes, stickyBatchSize, enableSwitch);
+ }
+
+ if (producedBytes >= stickyBatchSize && enableSwitch || producedBytes >= stickyBatchSize * 2) {
// We've produced enough to this partition, switch to next.
StickyPartitionInfo newPartitionInfo = new StickyPartitionInfo(nextPartition(cluster));
stickyPartitionInfo.set(newPartitionInfo);
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 4168ea68aa430..983426316df9e 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
@@ -213,6 +213,43 @@ private void setPartition(AppendCallbacks callbacks, int partition) {
callbacks.setPartition(partition);
}
+ /**
+ * Check if partition concurrently changed, or we need to complete previously disabled partition change.
+ *
+ * @param topic The topic
+ * @param topicInfo The topic info
+ * @param partitionInfo The built-in partitioner's partition info
+ * @param deque The partition queue
+ * @param nowMs The current time, in milliseconds
+ * @param cluster THe cluster metadata
+ * @return 'true' if partition changed and we need to get new partition info and retry,
+ * 'false' otherwise
+ */
+ private boolean partitionChanged(String topic,
+ TopicInfo topicInfo,
+ BuiltInPartitioner.StickyPartitionInfo partitionInfo,
+ Deque deque, long nowMs,
+ Cluster cluster) {
+ if (topicInfo.builtInPartitioner.isPartitionChanged(partitionInfo)) {
+ log.trace("Partition {} for topic {} switched by a concurrent append, retrying",
+ partitionInfo.partition(), topic);
+ return true;
+ }
+
+ // We might have disabled partition switch if the queue had incomplete batches.
+ // Check if all batches are full now and switch .
+ if (allBatchesFull(deque)) {
+ topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, 0, cluster, true);
+ if (topicInfo.builtInPartitioner.isPartitionChanged(partitionInfo)) {
+ log.trace("Completed previously disabled switch for topic {} partition {}, retrying",
+ topic, partitionInfo.partition());
+ return true;
+ }
+ }
+
+ return false;
+ }
+
/**
* Add a record to the accumulator, return the append result
*
@@ -275,14 +312,14 @@ public RecordAppendResult append(String topic,
Deque dq = topicInfo.batches.computeIfAbsent(effectivePartition, k -> new ArrayDeque<>());
synchronized (dq) {
// After taking the lock, validate that the partition hasn't changed and retry.
- if (topicInfo.builtInPartitioner.isPartitionChanged(partitionInfo)) {
- log.trace("Partition {} for topic {} switched by a concurrent append, retrying",
- partitionInfo.partition(), topic);
+ if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
continue;
- }
+
RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
if (appendResult != null) {
- topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster);
+ // If queue has incomplete batches we disable switch (see comments in updatePartitionInfo).
+ boolean enableSwitch = allBatchesFull(dq);
+ topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
return appendResult;
}
}
@@ -297,21 +334,26 @@ public RecordAppendResult append(String topic,
byte maxUsableMagic = apiVersions.maxUsableProduceMagic();
int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));
log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, partition, maxTimeToBlock);
+ // This call may block if we exhausted buffer space.
buffer = free.allocate(size, maxTimeToBlock);
+ // Update the current time in case the buffer allocation blocked above.
+ // NOTE: getting time may be expensive, so calling it under a lock
+ // should be avoided.
+ nowMs = time.milliseconds();
}
synchronized (dq) {
// After taking the lock, validate that the partition hasn't changed and retry.
- if (topicInfo.builtInPartitioner.isPartitionChanged(partitionInfo)) {
- log.trace("Partition {} for topic {} switched by a concurrent append, retrying",
- partitionInfo.partition(), topic);
+ if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
continue;
- }
- RecordAppendResult appendResult = appendNewBatch(topic, effectivePartition, dq, timestamp, key, value, headers, callbacks, buffer);
+
+ RecordAppendResult appendResult = appendNewBatch(topic, effectivePartition, dq, timestamp, key, value, headers, callbacks, buffer, nowMs);
// Set buffer to null, so that deallocate doesn't return it back to free pool, since it's used in the batch.
if (appendResult.newBatchCreated)
buffer = null;
- topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster);
+ // If queue has incomplete batches we disable switch (see comments in updatePartitionInfo).
+ boolean enableSwitch = allBatchesFull(dq);
+ topicInfo.builtInPartitioner.updatePartitionInfo(partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
return appendResult;
}
}
@@ -333,6 +375,7 @@ public RecordAppendResult append(String topic,
* @param headers the Headers for the record
* @param callbacks The callbacks to execute
* @param buffer The buffer for the new batch
+ * @param nowMs The current time, in milliseconds
*/
private RecordAppendResult appendNewBatch(String topic,
int partition,
@@ -342,11 +385,10 @@ private RecordAppendResult appendNewBatch(String topic,
byte[] value,
Header[] headers,
AppendCallbacks callbacks,
- ByteBuffer buffer) {
+ ByteBuffer buffer,
+ long nowMs) {
assert partition != RecordMetadata.UNKNOWN_PARTITION;
- // Update the current time in case the buffer allocation blocked above.
- long nowMs = time.milliseconds();
RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
if (appendResult != null) {
// Somebody else found us a batch, return the one we waited for! Hopefully this doesn't happen often...
@@ -373,6 +415,15 @@ private MemoryRecordsBuilder recordsBuilder(ByteBuffer buffer, byte maxUsableMag
}
/**
+ * Check if all batches in the queue are full.
+ */
+ private boolean allBatchesFull(Deque deque) {
+ // Only the last batch may be incomplete, so we just check that.
+ ProducerBatch last = deque.peekLast();
+ return last == null || last.isFull();
+ }
+
+ /**
* Try to append to a ProducerBatch.
*
* If it is full, we return null and a new batch is created. We also close the batch for record appends to free up
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
index 5aab62eaf225a..de5a6ced41c85 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
@@ -1290,7 +1290,7 @@ public void handleResponse(AbstractResponse response) {
} else if (error == Errors.NOT_COORDINATOR || error == Errors.COORDINATOR_NOT_AVAILABLE) {
lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
reenqueue();
- } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
+ } else if (error.exception() instanceof RetriableException) {
reenqueue();
} else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
error == Errors.CLUSTER_AUTHORIZATION_FAILED) {
@@ -1347,7 +1347,7 @@ public void handleResponse(AbstractResponse response) {
maybeOverrideRetryBackoffMs();
reenqueue();
return;
- } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ } else if (error.exception() instanceof RetriableException) {
reenqueue();
return;
} else if (error == Errors.INVALID_PRODUCER_EPOCH || error == Errors.PRODUCER_FENCED) {
@@ -1467,7 +1467,7 @@ public void handleResponse(AbstractResponse response) {
}
result.done();
log.info("Discovered {} coordinator {}", coordinatorType.toString().toLowerCase(Locale.ROOT), node);
- } else if (error == Errors.COORDINATOR_NOT_AVAILABLE) {
+ } else if (error.exception() instanceof RetriableException) {
reenqueue();
} else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
fatalError(error.exception());
@@ -1515,7 +1515,7 @@ public void handleResponse(AbstractResponse response) {
} else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) {
lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
reenqueue();
- } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
+ } else if (error.exception() instanceof RetriableException) {
reenqueue();
} else if (error == Errors.INVALID_PRODUCER_EPOCH || error == Errors.PRODUCER_FENCED) {
// We could still receive INVALID_PRODUCER_EPOCH from old versioned transaction coordinator,
@@ -1572,7 +1572,7 @@ public void handleResponse(AbstractResponse response) {
} else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) {
lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
reenqueue();
- } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
+ } else if (error.exception() instanceof RetriableException) {
reenqueue();
} else if (error == Errors.UNKNOWN_PRODUCER_ID || error == Errors.INVALID_PRODUCER_ID_MAPPING) {
abortableErrorIfPossible(error.exception());
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 7d3f6f08a0139..84b77ef5f40db 100644
--- a/clients/src/main/java/org/apache/kafka/common/Cluster.java
+++ b/clients/src/main/java/org/apache/kafka/common/Cluster.java
@@ -252,7 +252,12 @@ public Node nodeById(int id) {
*/
public Optional nodeIfOnline(TopicPartition partition, int id) {
Node node = nodeById(id);
- if (node != null && !Arrays.asList(partition(partition).offlineReplicas()).contains(node)) {
+ PartitionInfo partitionInfo = partition(partition);
+
+ if (node != null && partitionInfo != null &&
+ !Arrays.asList(partitionInfo.offlineReplicas()).contains(node) &&
+ Arrays.asList(partitionInfo.replicas()).contains(node)) {
+
return Optional.of(node);
} else {
return Optional.empty();
diff --git a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
index 0b90da8f800a6..8b7a9649c2c43 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
@@ -36,6 +36,8 @@ public class BrokerSecurityConfigs {
public static final String SASL_SERVER_CALLBACK_HANDLER_CLASS = "sasl.server.callback.handler.class";
public static final String SSL_PRINCIPAL_MAPPING_RULES_CONFIG = "ssl.principal.mapping.rules";
public static final String CONNECTIONS_MAX_REAUTH_MS = "connections.max.reauth.ms";
+ public static final int DEFAULT_SASL_SERVER_MAX_RECEIVE_SIZE = 524288;
+ public static final String SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG = "sasl.server.max.receive.size";
public static final String PRINCIPAL_BUILDER_CLASS_DOC = "The fully qualified name of a class that implements the " +
"KafkaPrincipalBuilder interface, which is used to build the KafkaPrincipal object used during " +
@@ -89,4 +91,8 @@ public class BrokerSecurityConfigs {
+ "The broker will disconnect any such connection that is not re-authenticated within the session lifetime and that is then subsequently "
+ "used for any purpose other than re-authentication. Configuration names can optionally be prefixed with listener prefix and SASL "
+ "mechanism name in lower-case. For example, listener.name.sasl_ssl.oauthbearer.connections.max.reauth.ms=3600000";
+
+ public static final String SASL_SERVER_MAX_RECEIVE_SIZE_DOC = "The maximum receive size allowed before and during initial SASL authentication." +
+ " Default receive size is 512KB. GSSAPI limits requests to 64K, but we allow upto 512KB by default for custom SASL mechanisms. In practice," +
+ " PLAIN, SCRAM and OAUTH mechanisms can use much smaller limits.";
}
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/ConcurrentTransactionsException.java b/clients/src/main/java/org/apache/kafka/common/errors/ConcurrentTransactionsException.java
index 6ad6b8a3ddb4c..118b4de50aadf 100644
--- a/clients/src/main/java/org/apache/kafka/common/errors/ConcurrentTransactionsException.java
+++ b/clients/src/main/java/org/apache/kafka/common/errors/ConcurrentTransactionsException.java
@@ -16,7 +16,7 @@
*/
package org.apache.kafka.common.errors;
-public class ConcurrentTransactionsException extends ApiException {
+public class ConcurrentTransactionsException extends RetriableException {
private static final long serialVersionUID = 1L;
public ConcurrentTransactionsException(final String message) {
diff --git a/clients/src/main/java/org/apache/kafka/common/internals/Topic.java b/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
index fbf491bab9800..fe4a01c52a635 100644
--- a/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
+++ b/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
@@ -28,8 +28,11 @@ public class Topic {
public static final String GROUP_METADATA_TOPIC_NAME = "__consumer_offsets";
public static final String TRANSACTION_STATE_TOPIC_NAME = "__transaction_state";
- public static final String METADATA_TOPIC_NAME = "__cluster_metadata";
- public static final TopicPartition METADATA_TOPIC_PARTITION = new TopicPartition(METADATA_TOPIC_NAME, 0);
+ public static final String CLUSTER_METADATA_TOPIC_NAME = "__cluster_metadata";
+ public static final TopicPartition CLUSTER_METADATA_TOPIC_PARTITION = new TopicPartition(
+ CLUSTER_METADATA_TOPIC_NAME,
+ 0
+ );
public static final String LEGAL_CHARS = "[a-zA-Z0-9._-]";
private static final Set INTERNAL_TOPICS = Collections.unmodifiableSet(
@@ -43,17 +46,31 @@ public static void validate(String topic) {
});
}
- public static void validate(String name, String logPrefix, Consumer throwableConsumer) {
+ private static String detectInvalidTopic(String name) {
if (name.isEmpty())
- throwableConsumer.accept(logPrefix + " is illegal, it can't be empty");
- if (".".equals(name) || "..".equals(name))
- throwableConsumer.accept(logPrefix + " cannot be \".\" or \"..\"");
+ return "the empty string is not allowed";
+ if (".".equals(name))
+ return "'.' is not allowed";
+ if ("..".equals(name))
+ return "'..' is not allowed";
if (name.length() > MAX_NAME_LENGTH)
- throwableConsumer.accept(logPrefix + " is illegal, it can't be longer than " + MAX_NAME_LENGTH +
- " characters, " + logPrefix + ": " + name);
+ return "the length of '" + name + "' is longer than the max allowed length " + MAX_NAME_LENGTH;
if (!containsValidPattern(name))
- throwableConsumer.accept(logPrefix + " \"" + name + "\" is illegal, it contains a character other than " +
- "ASCII alphanumerics, '.', '_' and '-'");
+ return "'" + name + "' contains one or more characters other than " +
+ "ASCII alphanumerics, '.', '_' and '-'";
+ return null;
+ }
+
+ public static boolean isValid(String name) {
+ String reasonInvalid = detectInvalidTopic(name);
+ return reasonInvalid == null;
+ }
+
+ public static void validate(String name, String logPrefix, Consumer throwableConsumer) {
+ String reasonInvalid = detectInvalidTopic(name);
+ if (reasonInvalid != null) {
+ throwableConsumer.accept(logPrefix + " is invalid: " + reasonInvalid);
+ }
}
public static boolean isInternal(String topic) {
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
index c6b8574186a88..09b7c05c8f283 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java
@@ -91,7 +91,10 @@ public long windowSize(MetricConfig config, long now) {
if (numFullWindows < minFullWindows)
totalElapsedTimeMs += (minFullWindows - numFullWindows) * config.timeWindowMs();
- return totalElapsedTimeMs;
+ // If window size is being calculated at the exact beginning of the window with no prior samples, the window size
+ // will result in a value of 0. Calculation of rate over a window is size 0 is undefined, hence, we assume the
+ // minimum window size to be at least 1ms.
+ return Math.max(totalElapsedTimeMs, 1);
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
index 628c9407cc2dd..f727bd18e5fb9 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
@@ -231,7 +231,7 @@ private static String toHtml() {
b.append("| Name | \n");
b.append("Key | \n");
b.append("");
- for (ApiKeys key : zkBrokerApis()) {
+ for (ApiKeys key : clientApis()) {
b.append("\n");
b.append("");
b.append("" + key.name + "");
@@ -271,15 +271,22 @@ public static EnumSet controllerApis() {
return apisForListener(ApiMessageType.ListenerType.CONTROLLER);
}
+ public static EnumSet clientApis() {
+ List apis = Arrays.stream(ApiKeys.values())
+ .filter(apiKey -> apiKey.inScope(ApiMessageType.ListenerType.ZK_BROKER) || apiKey.inScope(ApiMessageType.ListenerType.BROKER))
+ .collect(Collectors.toList());
+ return EnumSet.copyOf(apis);
+ }
+
public static EnumSet apisForListener(ApiMessageType.ListenerType listener) {
return APIS_BY_LISTENER.get(listener);
}
private static EnumSet filterApisForListener(ApiMessageType.ListenerType listener) {
- List controllerApis = Arrays.stream(ApiKeys.values())
- .filter(apiKey -> apiKey.messageType.listeners().contains(listener))
+ List apis = Arrays.stream(ApiKeys.values())
+ .filter(apiKey -> apiKey.inScope(listener))
.collect(Collectors.toList());
- return EnumSet.copyOf(controllerApis);
+ return EnumSet.copyOf(apis);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java b/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java
index bd0925d6db358..f643f5b5779b1 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java
@@ -54,8 +54,15 @@ public double readDouble() {
}
@Override
- public void readArray(byte[] arr) {
+ public byte[] readArray(int size) {
+ int remaining = buf.remaining();
+ if (size > remaining) {
+ throw new RuntimeException("Error reading byte array of " + size + " byte(s): only " + remaining +
+ " byte(s) available");
+ }
+ byte[] arr = new byte[size];
buf.get(arr);
+ return arr;
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/DataInputStreamReadable.java b/clients/src/main/java/org/apache/kafka/common/protocol/DataInputStreamReadable.java
deleted file mode 100644
index 70ed52d6a02f6..0000000000000
--- a/clients/src/main/java/org/apache/kafka/common/protocol/DataInputStreamReadable.java
+++ /dev/null
@@ -1,139 +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.protocol;
-
-import org.apache.kafka.common.utils.ByteUtils;
-
-import java.io.Closeable;
-import java.io.DataInputStream;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-public class DataInputStreamReadable implements Readable, Closeable {
- protected final DataInputStream input;
-
- public DataInputStreamReadable(DataInputStream input) {
- this.input = input;
- }
-
- @Override
- public byte readByte() {
- try {
- return input.readByte();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public short readShort() {
- try {
- return input.readShort();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public int readInt() {
- try {
- return input.readInt();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public long readLong() {
- try {
- return input.readLong();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public double readDouble() {
- try {
- return input.readDouble();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void readArray(byte[] arr) {
- try {
- input.readFully(arr);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public int readUnsignedVarint() {
- try {
- return ByteUtils.readUnsignedVarint(input);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public ByteBuffer readByteBuffer(int length) {
- byte[] arr = new byte[length];
- readArray(arr);
- return ByteBuffer.wrap(arr);
- }
-
- @Override
- public int readVarint() {
- try {
- return ByteUtils.readVarint(input);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public long readVarlong() {
- try {
- return ByteUtils.readVarlong(input);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public int remaining() {
- try {
- return input.available();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void close() {
- try {
- input.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
-}
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 2ca42bafcfb86..c220bbcde43cb 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
@@ -132,6 +132,8 @@
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
import java.util.function.Function;
/**
@@ -469,7 +471,8 @@ public static Errors forCode(short code) {
* If there are multiple matches in the class hierarchy, the first match starting from the bottom is used.
*/
public static Errors forException(Throwable t) {
- Class clazz = t.getClass();
+ Throwable cause = maybeUnwrapException(t);
+ Class clazz = cause.getClass();
while (clazz != null) {
Errors error = classToError.get(clazz);
if (error != null)
@@ -479,6 +482,22 @@ public static Errors forException(Throwable t) {
return UNKNOWN_SERVER_ERROR;
}
+ /**
+ * Check if a Throwable is a commonly wrapped exception type (e.g. `CompletionException`) and return
+ * the cause if so. This is useful to handle cases where exceptions may be raised from a future or a
+ * completion stage (as might be the case for requests sent to the controller in `ControllerApis`).
+ *
+ * @param t The Throwable to check
+ * @return The throwable itself or its cause if it is an instance of a commonly wrapped exception type
+ */
+ public static Throwable maybeUnwrapException(Throwable t) {
+ if (t instanceof CompletionException || t instanceof ExecutionException) {
+ return t.getCause();
+ } else {
+ return t;
+ }
+ }
+
private static String toHtml() {
final StringBuilder b = new StringBuilder();
b.append("\n");
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
index a75eb0661d102..2bc0fc6efef7d 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
@@ -133,7 +133,7 @@ public static String toHtml() {
b.append("\n");
schemaToFieldTableHtml(ResponseHeaderData.SCHEMAS[i], b);
}
- for (ApiKeys key : ApiKeys.zkBrokerApis()) {
+ for (ApiKeys key : ApiKeys.clientApis()) {
// Key
b.append(" |