filteredProducerIds() {
+ return filteredProducerIds;
+ }
+
+ @Override
+ public String toString() {
+ return "ListTransactionsOptions(" +
+ "filteredStates=" + filteredStates +
+ ", filteredProducerIds=" + filteredProducerIds +
+ ", timeoutMs=" + timeoutMs +
+ ')';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
new file mode 100644
index 0000000000000..4b9d4eee1beeb
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsResult.java
@@ -0,0 +1,125 @@
+/*
+ * 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.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * The result of the {@link Admin#listTransactions()} call.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+ private final KafkaFutureImpl>>> future;
+
+ ListTransactionsResult(KafkaFutureImpl>>> future) {
+ this.future = future;
+ }
+
+ /**
+ * Get all transaction listings. If any of the underlying requests fail, then the future
+ * returned from this method will also fail with the first encountered error.
+ *
+ * @return A future containing the collection of transaction listings. The future completes
+ * when all transaction listings are available and fails after any non-retriable error.
+ */
+ public KafkaFuture> all() {
+ return allByBrokerId().thenApply(map -> {
+ List allListings = new ArrayList<>();
+ for (Collection listings : map.values()) {
+ allListings.addAll(listings);
+ }
+ return allListings;
+ });
+ }
+
+ /**
+ * Get a future which returns a map containing the underlying listing future for each broker
+ * in the cluster. This is useful, for example, if a partial listing of transactions is
+ * sufficient, or if you want more granular error details.
+ *
+ * @return A future containing a map of futures by broker which complete individually when
+ * their respective transaction listings are available. The top-level future returned
+ * from this method may fail if the admin client is unable to lookup the available
+ * brokers in the cluster.
+ */
+ public KafkaFuture>>> byBrokerId() {
+ KafkaFutureImpl>>> result = new KafkaFutureImpl<>();
+ future.whenComplete((brokerFutures, exception) -> {
+ if (brokerFutures != null) {
+ Map>> brokerFuturesCopy =
+ new HashMap<>(brokerFutures.size());
+ brokerFuturesCopy.putAll(brokerFutures);
+ result.complete(brokerFuturesCopy);
+ } else {
+ result.completeExceptionally(exception);
+ }
+ });
+ return result;
+ }
+
+ /**
+ * Get all transaction listings in a map which is keyed by the ID of respective broker
+ * that is currently managing them. If any of the underlying requests fail, then the future
+ * returned from this method will also fail with the first encountered error.
+ *
+ * @return A future containing a map from the broker ID to the transactions hosted by that
+ * broker respectively. This future completes when all transaction listings are
+ * available and fails after any non-retriable error.
+ */
+ public KafkaFuture>> allByBrokerId() {
+ KafkaFutureImpl>> allFuture = new KafkaFutureImpl<>();
+ Map> allListingsMap = new HashMap<>();
+
+ future.whenComplete((map, topLevelException) -> {
+ if (topLevelException != null) {
+ allFuture.completeExceptionally(topLevelException);
+ return;
+ }
+
+ Set remainingResponses = new HashSet<>(map.keySet());
+ map.forEach((brokerId, future) -> {
+ future.whenComplete((listings, brokerException) -> {
+ if (brokerException != null) {
+ allFuture.completeExceptionally(brokerException);
+ } else if (!allFuture.isDone()) {
+ allListingsMap.put(brokerId, listings);
+ remainingResponses.remove(brokerId);
+
+ if (remainingResponses.isEmpty()) {
+ allFuture.complete(allListingsMap);
+ }
+ }
+ });
+ });
+ });
+
+ return allFuture;
+ }
+
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/TransactionListing.java b/clients/src/main/java/org/apache/kafka/clients/admin/TransactionListing.java
new file mode 100644
index 0000000000000..e8e7eb6143baf
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/TransactionListing.java
@@ -0,0 +1,74 @@
+/*
+ * 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.annotation.InterfaceStability;
+
+import java.util.Objects;
+
+@InterfaceStability.Evolving
+public class TransactionListing {
+ private final String transactionalId;
+ private final long producerId;
+ private final TransactionState transactionState;
+
+ public TransactionListing(
+ String transactionalId,
+ long producerId,
+ TransactionState transactionState
+ ) {
+ this.transactionalId = transactionalId;
+ this.producerId = producerId;
+ this.transactionState = transactionState;
+ }
+
+ public String transactionalId() {
+ return transactionalId;
+ }
+
+ public long producerId() {
+ return producerId;
+ }
+
+ public TransactionState state() {
+ return transactionState;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ TransactionListing that = (TransactionListing) o;
+ return producerId == that.producerId &&
+ Objects.equals(transactionalId, that.transactionalId) &&
+ transactionState == that.transactionState;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(transactionalId, producerId, transactionState);
+ }
+
+ @Override
+ public String toString() {
+ return "TransactionListing(" +
+ "transactionalId='" + transactionalId + '\'' +
+ ", producerId=" + producerId +
+ ", transactionState=" + transactionState +
+ ')';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java
index 09dc7be55162f..90fda93a5fa4e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java
@@ -31,7 +31,6 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -52,17 +51,20 @@ public AbortTransactionHandler(
this.lookupStrategy = new PartitionLeaderStrategy(logContext);
}
+ public static AdminApiFuture.SimpleAdminApiFuture newFuture(
+ Set topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(topicPartitions);
+ }
+
@Override
public String apiName() {
return "abortTransaction";
}
@Override
- public Keys initializeKeys() {
- return Keys.dynamicMapped(
- Collections.singleton(abortSpec.topicPartition()),
- lookupStrategy
- );
+ public AdminApiLookupStrategy lookupStrategy() {
+ return lookupStrategy;
}
@Override
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 45ab667db917c..a434f789e04e1 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
@@ -16,15 +16,14 @@
*/
package org.apache.kafka.clients.admin.internals;
-import org.apache.kafka.clients.admin.internals.AdminApiHandler.Keys;
import org.apache.kafka.common.errors.DisconnectException;
-import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -35,6 +34,8 @@
import java.util.OptionalInt;
import java.util.Set;
import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.stream.Collectors;
/**
* The `KafkaAdminClient`'s internal `Call` primitive is not a good fit for multi-stage
@@ -79,8 +80,7 @@ public class AdminApiDriver {
private final long retryBackoffMs;
private final long deadlineMs;
private final AdminApiHandler handler;
- private final Keys keys;
- private final Map> futures;
+ private final AdminApiFuture future;
private final BiMultimap lookupMap = new BiMultimap<>();
private final BiMultimap fulfillmentMap = new BiMultimap<>();
@@ -88,38 +88,23 @@ public class AdminApiDriver {
public AdminApiDriver(
AdminApiHandler handler,
+ AdminApiFuture future,
long deadlineMs,
long retryBackoffMs,
LogContext logContext
) {
this.handler = handler;
+ this.future = future;
this.deadlineMs = deadlineMs;
this.retryBackoffMs = retryBackoffMs;
this.log = logContext.logger(AdminApiDriver.class);
- this.futures = new HashMap<>();
- this.keys = initializeKeys(handler);
- }
-
- private Keys initializeKeys(AdminApiHandler handler) {
- Keys keys = handler.initializeKeys();
-
- keys.staticKeys.forEach((key, brokerId) -> {
- futures.put(key, new KafkaFutureImpl<>());
- map(key, brokerId);
- });
-
- keys.dynamicKeys.forEach(key -> {
- futures.put(key, new KafkaFutureImpl<>());
- lookupMap.put(keys.lookupStrategy.lookupScope(key), key);
- });
-
- return keys;
+ retryLookup(future.lookupKeys());
}
/**
* Associate a key with a brokerId. This is called after a response in the Lookup
- * stage reveals the mapping (e.g. when the `FindCoordinator` tells us the the
- * group coordinator for a specific consumer group).
+ * stage reveals the mapping (e.g. when the `FindCoordinator` tells us the group
+ * coordinator for a specific consumer group).
*/
private void map(K key, Integer brokerId) {
lookupMap.remove(key);
@@ -131,26 +116,30 @@ private void map(K key, Integer brokerId) {
* back to the Lookup stage, which will allow us to attempt lookup again.
*/
private void unmap(K key) {
- if (!keys.dynamicKeys.contains(key)) {
- throw new IllegalStateException("Attempt to unmap key " + key + " which is not dynamically mapped");
- }
-
fulfillmentMap.remove(key);
- lookupMap.put(keys.lookupStrategy.lookupScope(key), key);
+
+ ApiRequestScope lookupScope = handler.lookupStrategy().lookupScope(key);
+ OptionalInt destinationBrokerId = lookupScope.destinationBrokerId();
+
+ if (destinationBrokerId.isPresent()) {
+ fulfillmentMap.put(new FulfillmentScope(destinationBrokerId.getAsInt()), key);
+ } else {
+ lookupMap.put(handler.lookupStrategy().lookupScope(key), key);
+ }
}
- private void clear(K key) {
- lookupMap.remove(key);
- fulfillmentMap.remove(key);
+ private void clear(Collection keys) {
+ keys.forEach(key -> {
+ lookupMap.remove(key);
+ fulfillmentMap.remove(key);
+ });
}
OptionalInt keyToBrokerId(K key) {
Optional scope = fulfillmentMap.getKey(key);
- if (scope.isPresent()) {
- return OptionalInt.of(scope.get().destinationBrokerId);
- } else {
- return OptionalInt.empty();
- }
+ return scope
+ .map(fulfillmentScope -> OptionalInt.of(fulfillmentScope.destinationBrokerId))
+ .orElseGet(OptionalInt::empty);
}
/**
@@ -158,27 +147,39 @@ OptionalInt keyToBrokerId(K key) {
* the key will be taken out of both the Lookup and Fulfillment stages so that request
* are not retried.
*/
- private void completeExceptionally(K key, Throwable t) {
- KafkaFutureImpl future = futures.get(key);
- if (future == null) {
- log.warn("Attempt to complete future for {}, which was not requested", key);
- } else {
- clear(key);
- future.completeExceptionally(t);
+ private void completeExceptionally(Map errors) {
+ if (!errors.isEmpty()) {
+ future.completeExceptionally(errors);
+ clear(errors.keySet());
}
}
+ private void completeLookupExceptionally(Map errors) {
+ if (!errors.isEmpty()) {
+ future.completeLookupExceptionally(errors);
+ clear(errors.keySet());
+ }
+ }
+
+ private void retryLookup(Collection keys) {
+ keys.forEach(this::unmap);
+ }
+
/**
- * Complete the future associated with the given key. After is called, the key will
+ * Complete the future associated with the given key. After this is called, all keys will
* be taken out of both the Lookup and Fulfillment stages so that request are not retried.
*/
- private void complete(K key, V value) {
- KafkaFutureImpl future = futures.get(key);
- if (future == null) {
- log.warn("Attempt to complete future for {}, which was not requested", key);
- } else {
- clear(key);
- future.complete(value);
+ private void complete(Map values) {
+ if (!values.isEmpty()) {
+ future.complete(values);
+ clear(values.keySet());
+ }
+ }
+
+ private void completeLookup(Map brokerIdMapping) {
+ if (!brokerIdMapping.isEmpty()) {
+ future.completeLookup(brokerIdMapping);
+ brokerIdMapping.forEach(this::map);
}
}
@@ -197,13 +198,6 @@ public List> poll() {
return requests;
}
- /**
- * Get a map of the futures that are awaiting completion.
- */
- public Map> futures() {
- return futures;
- }
-
/**
* Callback that is invoked when a `Call` returns a response successfully.
*/
@@ -221,16 +215,18 @@ public void onResponse(
spec.keys,
response
);
- result.completedKeys.forEach(this::complete);
- result.failedKeys.forEach(this::completeExceptionally);
- result.unmappedKeys.forEach(this::unmap);
+ complete(result.completedKeys);
+ completeExceptionally(result.failedKeys);
+ retryLookup(result.unmappedKeys);
} else {
- AdminApiLookupStrategy.LookupResult result = keys.lookupStrategy.handleResponse(
+ AdminApiLookupStrategy.LookupResult result = handler.lookupStrategy().handleResponse(
spec.keys,
response
);
- result.failedKeys.forEach(this::completeExceptionally);
- result.mappedKeys.forEach(this::map);
+
+ result.completedKeys.forEach(lookupMap::remove);
+ completeLookup(result.mappedKeys);
+ completeLookupExceptionally(result.failedKeys);
}
}
@@ -248,13 +244,23 @@ public void onFailure(
"Will attempt retry", spec.request);
// After a disconnect, we want the driver to attempt to lookup the key
- // again (if the key is dynamically mapped). This gives us a chance to
- // find a new coordinator or partition leader for example.
- spec.keys.stream()
- .filter(keys.dynamicKeys::contains)
- .forEach(this::unmap);
+ // again. This gives us a chance to find a new coordinator or partition
+ // leader for example.
+ Set keysToUnmap = spec.keys.stream()
+ .filter(future.lookupKeys()::contains)
+ .collect(Collectors.toSet());
+ retryLookup(keysToUnmap);
} else {
- spec.keys.forEach(key -> completeExceptionally(key, t));
+ Map errors = spec.keys.stream().collect(Collectors.toMap(
+ Function.identity(),
+ key -> t
+ ));
+
+ if (spec.scope instanceof FulfillmentScope) {
+ completeExceptionally(errors);
+ } else {
+ completeLookupExceptionally(errors);
+ }
}
}
@@ -303,13 +309,11 @@ private void collectRequests(
}
private void collectLookupRequests(List> requests) {
- if (!keys.dynamicKeys.isEmpty()) {
- collectRequests(
- requests,
- lookupMap,
- (keys, scope) -> this.keys.lookupStrategy.buildRequest(keys)
- );
- }
+ collectRequests(
+ requests,
+ lookupMap,
+ (keys, scope) -> handler.lookupStrategy().buildRequest(keys)
+ );
}
private void collectFulfillmentRequests(List> requests) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiFuture.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiFuture.java
new file mode 100644
index 0000000000000..00fabb3d1cb64
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiFuture.java
@@ -0,0 +1,125 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public interface AdminApiFuture {
+
+ /**
+ * The initial set of lookup keys. Although this will usually match the fulfillment
+ * keys, it does not necessarily have to. For example, in the case of
+ * {@link AllBrokersStrategy.AllBrokersFuture},
+ * we use the lookup phase in order to discover the set of keys that will be searched
+ * during the fulfillment phase.
+ *
+ * @return non-empty set of initial lookup keys
+ */
+ Set lookupKeys();
+
+ /**
+ * Complete the futures associated with the given keys.
+ *
+ * @param values the completed keys with their respective values
+ */
+ void complete(Map values);
+
+ /**
+ * Invoked when lookup of a set of keys succeeds.
+ *
+ * @param brokerIdMapping the discovered mapping from key to the respective brokerId that will
+ * handle the fulfillment request
+ */
+ default void completeLookup(Map brokerIdMapping) {
+ }
+
+ /**
+ * Invoked when lookup fails with a fatal error on a set of keys.
+ *
+ * @param lookupErrors the set of keys that failed lookup with their respective errors
+ */
+ default void completeLookupExceptionally(Map lookupErrors) {
+ completeExceptionally(lookupErrors);
+ }
+
+ /**
+ * Complete the futures associated with the given keys exceptionally.
+ *
+ * @param errors the failed keys with their respective errors
+ */
+ void completeExceptionally(Map errors);
+
+ static SimpleAdminApiFuture forKeys(Set keys) {
+ return new SimpleAdminApiFuture<>(keys);
+ }
+
+ /**
+ * This class can be used when the set of keys is known ahead of time.
+ */
+ class SimpleAdminApiFuture implements AdminApiFuture {
+ private final Map> futures;
+
+ public SimpleAdminApiFuture(Set keys) {
+ this.futures = keys.stream().collect(Collectors.toMap(
+ Function.identity(),
+ k -> new KafkaFutureImpl<>()
+ ));
+ }
+
+ @Override
+ public Set lookupKeys() {
+ return futures.keySet();
+ }
+
+ @Override
+ public void complete(Map values) {
+ values.forEach(this::complete);
+ }
+
+ private void complete(K key, V value) {
+ futureOrThrow(key).complete(value);
+ }
+
+ @Override
+ public void completeExceptionally(Map errors) {
+ errors.forEach(this::completeExceptionally);
+ }
+
+ private void completeExceptionally(K key, Throwable t) {
+ futureOrThrow(key).completeExceptionally(t);
+ }
+
+ private KafkaFutureImpl futureOrThrow(K key) {
+ KafkaFutureImpl future = futures.get(key);
+ if (future == null) {
+ throw new IllegalArgumentException("Attempt to complete future for " + key +
+ ", which was not requested");
+ } else {
+ return future;
+ }
+ }
+
+ public Map> all() {
+ return futures;
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
index 754ae89451fe0..db3c68e5cc2b1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
@@ -18,16 +18,12 @@
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
-import org.apache.kafka.common.utils.Utils;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import static java.util.Objects.requireNonNull;
-
public interface AdminApiHandler {
/**
@@ -35,19 +31,6 @@ public interface AdminApiHandler {
*/
String apiName();
- /**
- * Initialize the set of keys required to handle this API and how the driver
- * should map them to the broker that will handle the request for these keys.
- *
- * Two mapping types are supported:
- *
- * - Static mapping: when the brokerId is known ahead of time
- * - Dynamic mapping: when the brokerId must be discovered dynamically
- *
- * @return the key mappings
- */
- Keys initializeKeys();
-
/**
* Build the request. The set of keys are derived by {@link AdminApiDriver}
* during the lookup stage as the set of keys which all map to the same
@@ -82,44 +65,13 @@ public interface AdminApiHandler {
*/
ApiResult handleResponse(int brokerId, Set keys, AbstractResponse response);
- class Keys {
- public final Map staticKeys;
- public final Set dynamicKeys;
- public final AdminApiLookupStrategy lookupStrategy;
-
- public Keys(
- Map staticKeys,
- Set dynamicKeys,
- AdminApiLookupStrategy lookupStrategy
- ) {
- this.staticKeys = requireNonNull(staticKeys);
- this.dynamicKeys = requireNonNull(dynamicKeys);
- this.lookupStrategy = lookupStrategy;
-
- Set staticAndDynamicKeys = Utils.intersection(HashSet::new, staticKeys.keySet(), dynamicKeys);
- if (!staticAndDynamicKeys.isEmpty()) {
- throw new IllegalArgumentException("The following keys were configured both as dynamically " +
- "and statically mapped: " + staticAndDynamicKeys);
- }
-
- if (!dynamicKeys.isEmpty()) {
- requireNonNull(lookupStrategy);
- }
- }
-
- public static Keys staticMapped(
- Map staticKeys
- ) {
- return new Keys<>(staticKeys, Collections.emptySet(), null);
- }
-
- public static Keys dynamicMapped(
- Set dynamicKeys,
- AdminApiLookupStrategy lookupStrategy
- ) {
- return new Keys<>(Collections.emptyMap(), dynamicKeys, lookupStrategy);
- }
- }
+ /**
+ * Get the lookup strategy that is responsible for finding the brokerId
+ * which will handle each respective key.
+ *
+ * @return non-null lookup strategy
+ */
+ AdminApiLookupStrategy lookupStrategy();
class ApiResult {
public final Map completedKeys;
@@ -159,6 +111,14 @@ public static ApiResult unmapped(List keys) {
keys
);
}
+
+ public static ApiResult empty() {
+ return new ApiResult<>(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyList()
+ );
+ }
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiLookupStrategy.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiLookupStrategy.java
index 7969d534a9762..56c0837135772 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiLookupStrategy.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiLookupStrategy.java
@@ -20,6 +20,7 @@
import org.apache.kafka.common.requests.AbstractResponse;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -39,6 +40,11 @@ public interface AdminApiLookupStrategy {
* single key. This can be supported by returning a different scope object
* for each lookup key.
*
+ * Note that if the {@link ApiRequestScope#destinationBrokerId()} maps to
+ * a specific brokerId, then lookup will be skipped. See the use of
+ * {@link StaticBrokerStrategy} in {@link DescribeProducersHandler} for
+ * an example of this usage.
+ *
* @param key the lookup key
*
* @return request scope indicating how lookup requests can be batched together
@@ -76,13 +82,31 @@ public interface AdminApiLookupStrategy {
LookupResult handleResponse(Set keys, AbstractResponse response);
class LookupResult {
+ // This is the set of keys that have been completed by the lookup phase itself.
+ // The driver will not attempt lookup or fulfillment for completed keys.
+ public final List completedKeys;
+
+ // This is the set of keys that have been mapped to a specific broker for
+ // fulfillment of the API request.
public final Map mappedKeys;
+
+ // This is the set of keys that have encountered a fatal error during the lookup
+ // phase. The driver will not attempt lookup or fulfillment for failed keys.
public final Map failedKeys;
public LookupResult(
Map failedKeys,
Map mappedKeys
) {
+ this(Collections.emptyList(), failedKeys, mappedKeys);
+ }
+
+ public LookupResult(
+ List completedKeys,
+ Map failedKeys,
+ Map mappedKeys
+ ) {
+ this.completedKeys = Collections.unmodifiableList(completedKeys);
this.failedKeys = Collections.unmodifiableMap(failedKeys);
this.mappedKeys = Collections.unmodifiableMap(mappedKeys);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
index 6e834520c46f7..f1ccd9aeb5454 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java
@@ -38,7 +38,7 @@
* service thread (which also uses the NetworkClient).
*/
public class AdminMetadataManager {
- private Logger log;
+ private final Logger log;
/**
* The minimum amount of time that we should wait between subsequent
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java
new file mode 100644
index 0000000000000..56305db03d267
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategy.java
@@ -0,0 +1,212 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.internals.KafkaFutureImpl;
+import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.MetadataResponseData;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalInt;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * This class is used for use cases which require requests to be sent to all
+ * brokers in the cluster.
+ *
+ * This is a slightly degenerate case of a lookup strategy in the sense that
+ * the broker IDs are used as both the keys and values. Also, unlike
+ * {@link CoordinatorStrategy} and {@link PartitionLeaderStrategy}, we do not
+ * know the set of keys ahead of time: we require the initial lookup in order
+ * to discover what the broker IDs are. This is represented with a more complex
+ * type {@code Future>} in the admin API result type.
+ * For example, see {@link org.apache.kafka.clients.admin.ListTransactionsResult}.
+ */
+public class AllBrokersStrategy implements AdminApiLookupStrategy {
+ public static final BrokerKey ANY_BROKER = new BrokerKey(OptionalInt.empty());
+ public static final Set LOOKUP_KEYS = Collections.singleton(ANY_BROKER);
+ private static final ApiRequestScope SINGLE_REQUEST_SCOPE = new ApiRequestScope() {
+ };
+
+ private final Logger log;
+
+ public AllBrokersStrategy(
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(AllBrokersStrategy.class);
+ }
+
+ @Override
+ public ApiRequestScope lookupScope(BrokerKey key) {
+ return SINGLE_REQUEST_SCOPE;
+ }
+
+ @Override
+ public MetadataRequest.Builder buildRequest(Set keys) {
+ validateLookupKeys(keys);
+ // Send empty `Metadata` request. We are only interested in the brokers from the response
+ return new MetadataRequest.Builder(new MetadataRequestData());
+ }
+
+ @Override
+ public LookupResult handleResponse(Set keys, AbstractResponse abstractResponse) {
+ validateLookupKeys(keys);
+
+ MetadataResponse response = (MetadataResponse) abstractResponse;
+ MetadataResponseData.MetadataResponseBrokerCollection brokers = response.data().brokers();
+
+ if (brokers.isEmpty()) {
+ log.debug("Metadata response contained no brokers. Will backoff and retry");
+ return LookupResult.empty();
+ } else {
+ log.debug("Discovered all brokers {} to send requests to", brokers);
+ }
+
+ Map brokerKeys = brokers.stream().collect(Collectors.toMap(
+ broker -> new BrokerKey(OptionalInt.of(broker.nodeId())),
+ MetadataResponseData.MetadataResponseBroker::nodeId
+ ));
+
+ return new LookupResult<>(
+ Collections.singletonList(ANY_BROKER),
+ Collections.emptyMap(),
+ brokerKeys
+ );
+ }
+
+ private void validateLookupKeys(Set keys) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set: " + keys);
+ }
+ BrokerKey key = keys.iterator().next();
+ if (key != ANY_BROKER) {
+ throw new IllegalArgumentException("Unexpected key set: " + keys);
+ }
+ }
+
+ public static class BrokerKey {
+ public final OptionalInt brokerId;
+
+ public BrokerKey(OptionalInt brokerId) {
+ this.brokerId = brokerId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ BrokerKey that = (BrokerKey) o;
+ return Objects.equals(brokerId, that.brokerId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(brokerId);
+ }
+
+ @Override
+ public String toString() {
+ return "BrokerKey(" +
+ "brokerId=" + brokerId +
+ ')';
+ }
+ }
+
+ public static class AllBrokersFuture implements AdminApiFuture {
+ private final KafkaFutureImpl>> future = new KafkaFutureImpl<>();
+ private final Map> brokerFutures = new HashMap<>();
+
+ @Override
+ public Set lookupKeys() {
+ return LOOKUP_KEYS;
+ }
+
+ @Override
+ public void completeLookup(Map brokerMapping) {
+ brokerMapping.forEach((brokerKey, brokerId) -> {
+ if (brokerId != brokerKey.brokerId.orElse(-1)) {
+ throw new IllegalArgumentException("Invalid lookup mapping " + brokerKey + " -> " + brokerId);
+ }
+ brokerFutures.put(brokerId, new KafkaFutureImpl<>());
+ });
+ future.complete(brokerFutures);
+ }
+
+ @Override
+ public void completeLookupExceptionally(Map lookupErrors) {
+ if (!LOOKUP_KEYS.equals(lookupErrors.keySet())) {
+ throw new IllegalArgumentException("Unexpected keys among lookup errors: " + lookupErrors);
+ }
+ future.completeExceptionally(lookupErrors.get(ANY_BROKER));
+ }
+
+ @Override
+ public void complete(Map values) {
+ values.forEach(this::complete);
+ }
+
+ private void complete(AllBrokersStrategy.BrokerKey key, V value) {
+ if (key == ANY_BROKER) {
+ throw new IllegalArgumentException("Invalid attempt to complete with lookup key sentinel");
+ } else {
+ futureOrThrow(key).complete(value);
+ }
+ }
+
+ @Override
+ public void completeExceptionally(Map errors) {
+ errors.forEach(this::completeExceptionally);
+ }
+
+ private void completeExceptionally(AllBrokersStrategy.BrokerKey key, Throwable t) {
+ if (key == ANY_BROKER) {
+ future.completeExceptionally(t);
+ } else {
+ futureOrThrow(key).completeExceptionally(t);
+ }
+ }
+
+ public KafkaFutureImpl>> all() {
+ return future;
+ }
+
+ private KafkaFutureImpl futureOrThrow(BrokerKey key) {
+ if (!key.brokerId.isPresent()) {
+ throw new IllegalArgumentException("Attempt to complete with invalid key: " + key);
+ } else {
+ int brokerId = key.brokerId.getAsInt();
+ KafkaFutureImpl future = brokerFutures.get(brokerId);
+ if (future == null) {
+ throw new IllegalArgumentException("Attempt to complete with unknown broker id: " + brokerId);
+ } else {
+ return future;
+ }
+ }
+ }
+
+ }
+
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ApiRequestScope.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ApiRequestScope.java
index a216e76049408..0d27f38b29f37 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ApiRequestScope.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ApiRequestScope.java
@@ -33,7 +33,11 @@ public interface ApiRequestScope {
* Get the target broker ID that a request is intended for or
* empty if the request can be sent to any broker.
*
- * @return optional broker id
+ * Note that if the destination broker ID is present in the
+ * {@link ApiRequestScope} returned by {@link AdminApiLookupStrategy#lookupScope(Object)},
+ * then no lookup will be attempted.
+ *
+ * @return optional broker ID
*/
default OptionalInt destinationBrokerId() {
return OptionalInt.empty();
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java
index 2c62461d5c9f9..de1061f124dbe 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java
@@ -31,12 +31,13 @@
import org.apache.kafka.common.requests.DescribeProducersResponse;
import org.apache.kafka.common.utils.CollectionUtils;
import org.apache.kafka.common.utils.LogContext;
-import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
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.OptionalInt;
@@ -45,20 +46,28 @@
import java.util.stream.Collectors;
public class DescribeProducersHandler implements AdminApiHandler {
- private final LogContext logContext;
private final Logger log;
private final DescribeProducersOptions options;
- private final Set topicPartitions;
+ private final AdminApiLookupStrategy lookupStrategy;
public DescribeProducersHandler(
- Set topicPartitions,
DescribeProducersOptions options,
LogContext logContext
) {
- this.topicPartitions = Collections.unmodifiableSet(topicPartitions);
this.options = options;
this.log = logContext.logger(DescribeProducersHandler.class);
- this.logContext = logContext;
+
+ if (options.brokerId().isPresent()) {
+ this.lookupStrategy = new StaticBrokerStrategy<>(options.brokerId().getAsInt());
+ } else {
+ this.lookupStrategy = new PartitionLeaderStrategy(logContext);
+ }
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture newFuture(
+ Collection topicPartitions
+ ) {
+ return AdminApiFuture.forKeys(new HashSet<>(topicPartitions));
}
@Override
@@ -67,17 +76,8 @@ public String apiName() {
}
@Override
- public Keys initializeKeys() {
- if (options.brokerId().isPresent()) {
- // If the options indicate a specific broker, then we can skip the lookup step
- int destinationBrokerId = options.brokerId().getAsInt();
- Map staticMappedPartitions =
- Utils.initializeMap(topicPartitions, () -> destinationBrokerId);
- return Keys.staticMapped(staticMappedPartitions);
- } else {
- PartitionLeaderStrategy lookupStrategy = new PartitionLeaderStrategy(logContext);
- return Keys.dynamicMapped(topicPartitions, lookupStrategy);
- }
+ public AdminApiLookupStrategy lookupStrategy() {
+ return lookupStrategy;
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java
index edeaf1ccca41f..ac45f597f1ec8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java
@@ -42,17 +42,20 @@
import java.util.stream.Collectors;
public class DescribeTransactionsHandler implements AdminApiHandler {
- private final LogContext logContext;
private final Logger log;
- private final Set keys;
+ private final AdminApiLookupStrategy lookupStrategy;
public DescribeTransactionsHandler(
- Collection transactionalIds,
LogContext logContext
) {
- this.keys = buildKeySet(transactionalIds);
this.log = logContext.logger(DescribeTransactionsHandler.class);
- this.logContext = logContext;
+ this.lookupStrategy = new CoordinatorStrategy(logContext);
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture newFuture(
+ Collection transactionalIds
+ ) {
+ return AdminApiFuture.forKeys(buildKeySet(transactionalIds));
}
private static Set buildKeySet(Collection transactionalIds) {
@@ -67,8 +70,8 @@ public String apiName() {
}
@Override
- public Keys initializeKeys() {
- return Keys.dynamicMapped(keys, new CoordinatorStrategy(logContext));
+ public AdminApiLookupStrategy lookupStrategy() {
+ return lookupStrategy;
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java
new file mode 100644
index 0000000000000..d1b4ec2d5b8d6
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java
@@ -0,0 +1,130 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.clients.admin.ListTransactionsOptions;
+import org.apache.kafka.clients.admin.TransactionListing;
+import org.apache.kafka.clients.admin.TransactionState;
+import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
+import org.apache.kafka.common.message.ListTransactionsRequestData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ListTransactionsRequest;
+import org.apache.kafka.common.requests.ListTransactionsResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ListTransactionsHandler implements AdminApiHandler> {
+ private final Logger log;
+ private final ListTransactionsOptions options;
+ private final AllBrokersStrategy lookupStrategy;
+
+ public ListTransactionsHandler(
+ ListTransactionsOptions options,
+ LogContext logContext
+ ) {
+ this.options = options;
+ this.log = logContext.logger(ListTransactionsHandler.class);
+ this.lookupStrategy = new AllBrokersStrategy(logContext);
+ }
+
+ public static AllBrokersStrategy.AllBrokersFuture> newFuture() {
+ return new AllBrokersStrategy.AllBrokersFuture<>();
+ }
+
+ @Override
+ public String apiName() {
+ return "listTransactions";
+ }
+
+ @Override
+ public AdminApiLookupStrategy lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ @Override
+ public ListTransactionsRequest.Builder buildRequest(
+ int brokerId,
+ Set keys
+ ) {
+ ListTransactionsRequestData request = new ListTransactionsRequestData();
+ request.setProducerIdFilters(new ArrayList<>(options.filteredProducerIds()));
+ request.setStateFilters(options.filteredStates().stream()
+ .map(TransactionState::toString)
+ .collect(Collectors.toList()));
+ return new ListTransactionsRequest.Builder(request);
+ }
+
+ @Override
+ public ApiResult> handleResponse(
+ int brokerId,
+ Set keys,
+ AbstractResponse abstractResponse
+ ) {
+ AllBrokersStrategy.BrokerKey key = requireSingleton(keys, brokerId);
+
+ ListTransactionsResponse response = (ListTransactionsResponse) abstractResponse;
+ Errors error = Errors.forCode(response.data().errorCode());
+
+ if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ log.debug("The `ListTransactions` request sent to broker {} failed because the " +
+ "coordinator is still loading state. Will try again after backing off", brokerId);
+ return ApiResult.empty();
+ } else if (error == Errors.COORDINATOR_NOT_AVAILABLE) {
+ log.debug("The `ListTransactions` request sent to broker {} failed because the " +
+ "coordinator is shutting down", brokerId);
+ return ApiResult.failed(key, new CoordinatorNotAvailableException("ListTransactions " +
+ "request sent to broker " + brokerId + " failed because the coordinator is shutting down"));
+ } else if (error != Errors.NONE) {
+ log.error("The `ListTransactions` request sent to broker {} failed because of an " +
+ "unexpected error {}", brokerId, error);
+ return ApiResult.failed(key, error.exception("ListTransactions request " +
+ "sent to broker " + brokerId + " failed with an unexpected exception"));
+ } else {
+ List listings = response.data().transactionStates().stream()
+ .map(transactionState -> new TransactionListing(
+ transactionState.transactionalId(),
+ transactionState.producerId(),
+ TransactionState.parse(transactionState.transactionState())))
+ .collect(Collectors.toList());
+ return ApiResult.completed(key, listings);
+ }
+ }
+
+ private AllBrokersStrategy.BrokerKey requireSingleton(
+ Set keys,
+ int brokerId
+ ) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unexpected key set: " + keys);
+ }
+
+ AllBrokersStrategy.BrokerKey key = keys.iterator().next();
+ if (!key.brokerId.isPresent() || key.brokerId.getAsInt() != brokerId) {
+ throw new IllegalArgumentException("Unexpected broker key: " + key);
+ }
+
+ return key;
+ }
+
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/StaticBrokerStrategy.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/StaticBrokerStrategy.java
new file mode 100644
index 0000000000000..7b66537fe39dc
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/StaticBrokerStrategy.java
@@ -0,0 +1,65 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+
+import java.util.OptionalInt;
+import java.util.Set;
+
+/**
+ * This lookup strategy is used when we already know the destination broker ID
+ * and we have no need for an explicit lookup. By setting {@link ApiRequestScope#destinationBrokerId()}
+ * in the returned value for {@link #lookupScope(Object)}, the driver will
+ * skip the lookup.
+ */
+public class StaticBrokerStrategy implements AdminApiLookupStrategy {
+ private final SingleBrokerScope scope;
+
+ public StaticBrokerStrategy(int brokerId) {
+ this.scope = new SingleBrokerScope(brokerId);
+ }
+
+ @Override
+ public ApiRequestScope lookupScope(K key) {
+ return scope;
+ }
+
+ @Override
+ public AbstractRequest.Builder buildRequest(Set keys) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public LookupResult handleResponse(Set keys, AbstractResponse response) {
+ throw new UnsupportedOperationException();
+ }
+
+ private static class SingleBrokerScope implements ApiRequestScope {
+ private final int brokerId;
+
+ private SingleBrokerScope(int brokerId) {
+ this.brokerId = brokerId;
+ }
+
+ @Override
+ public OptionalInt destinationBrokerId() {
+ return OptionalInt.of(brokerId);
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerInterceptor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerInterceptor.java
index 6af47058e4e21..c04afccd8aaf9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerInterceptor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerInterceptor.java
@@ -65,7 +65,7 @@ public interface ConsumerInterceptor extends Configurable, AutoCloseable {
* @param records records to be consumed by the client or records returned by the previous interceptors in the list.
* @return records that are either modified by the interceptor or same as records passed to this method.
*/
- public ConsumerRecords onConsume(ConsumerRecords records);
+ ConsumerRecords onConsume(ConsumerRecords records);
/**
* This is called when offsets get committed.
@@ -74,10 +74,10 @@ public interface ConsumerInterceptor extends Configurable, AutoCloseable {
*
* @param offsets A map of offsets by partition with associated metadata
*/
- public void onCommit(Map offsets);
+ void onCommit(Map offsets);
/**
* This is called when interceptor is closed
*/
- public void close();
+ void close();
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
index 5798909927461..e111aa62599e4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractStickyAssignor.java
@@ -29,7 +29,6 @@
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
-import java.util.TreeMap;
import java.util.TreeSet;
import java.util.stream.Collectors;
import org.apache.kafka.common.TopicPartition;
@@ -40,6 +39,7 @@ public abstract class AbstractStickyAssignor extends AbstractPartitionAssignor {
private static final Logger log = LoggerFactory.getLogger(AbstractStickyAssignor.class);
public static final int DEFAULT_GENERATION = -1;
+ public int maxGeneration = DEFAULT_GENERATION;
private PartitionMovements partitionMovements;
@@ -77,12 +77,10 @@ public Map> assign(Map partitionsP
partitionsTransferringOwnership = new HashMap<>();
return constrainedAssign(partitionsPerTopic, consumerToOwnedPartitions);
} else {
- log.debug("Detected that all not consumers were subscribed to same set of topics, falling back to the "
+ log.debug("Detected that not all consumers were subscribed to same set of topics, falling back to the "
+ "general case assignment algorithm");
partitionsTransferringOwnership = null;
- // we don't need consumerToOwnedPartitions in general assign case
- consumerToOwnedPartitions = null;
- return generalAssign(partitionsPerTopic, subscriptions);
+ return generalAssign(partitionsPerTopic, subscriptions, consumerToOwnedPartitions);
}
}
@@ -95,7 +93,7 @@ private boolean allSubscriptionsEqual(Set allTopics,
Map> consumerToOwnedPartitions) {
Set membersWithOldGeneration = new HashSet<>();
Set membersOfCurrentHighestGeneration = new HashSet<>();
- int maxGeneration = DEFAULT_GENERATION;
+ boolean isAllSubscriptionsEqual = true;
Set subscribedTopics = new HashSet<>();
@@ -106,9 +104,9 @@ private boolean allSubscriptionsEqual(Set allTopics,
// initialize the subscribed topics set if this is the first subscription
if (subscribedTopics.isEmpty()) {
subscribedTopics.addAll(subscription.topics());
- } else if (!(subscription.topics().size() == subscribedTopics.size()
+ } else if (isAllSubscriptionsEqual && !(subscription.topics().size() == subscribedTopics.size()
&& subscribedTopics.containsAll(subscription.topics()))) {
- return false;
+ isAllSubscriptionsEqual = false;
}
MemberData memberData = memberData(subscription);
@@ -141,7 +139,7 @@ private boolean allSubscriptionsEqual(Set allTopics,
for (String consumer : membersWithOldGeneration) {
consumerToOwnedPartitions.get(consumer).clear();
}
- return true;
+ return isAllSubscriptionsEqual;
}
@@ -301,62 +299,6 @@ private Map> constrainedAssign(Map
return assignment;
}
- /**
- * get the unassigned partition list by computing the difference set of all sorted partitions
- * and sortedAssignedPartitions. If no assigned partitions, we'll just return all topic partitions.
- *
- * To compute the difference set, we use two pointers technique here:
- *
- * We loop through the all sorted topics, and then iterate all partitions the topic has,
- * compared with the ith element in sortedAssignedPartitions(i starts from 0):
- * - if not equal to the ith element, add to unassignedPartitions
- * - if equal to the the ith element, get next element from sortedAssignedPartitions
- *
- * @param totalPartitionsCount all partitions counts in this assignment
- * @param partitionsPerTopic the number of partitions for each subscribed topic.
- * @param sortedAssignedPartitions sorted partitions, all are included in the sortedPartitions
- * @return the partitions not yet assigned to any consumers
- */
- private List getUnassignedPartitions(int totalPartitionsCount,
- Map partitionsPerTopic,
- List sortedAssignedPartitions) {
- List sortedAllTopics = new ArrayList<>(partitionsPerTopic.keySet());
- // sort all topics first, then we can have sorted all topic partitions by adding partitions starting from 0
- Collections.sort(sortedAllTopics);
-
- if (sortedAssignedPartitions.isEmpty()) {
- // no assigned partitions means all partitions are unassigned partitions
- return getAllTopicPartitions(partitionsPerTopic, sortedAllTopics, totalPartitionsCount);
- }
-
- List unassignedPartitions = new ArrayList<>(totalPartitionsCount - sortedAssignedPartitions.size());
-
- Collections.sort(sortedAssignedPartitions, Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
-
- boolean shouldAddDirectly = false;
- Iterator sortedAssignedPartitionsIter = sortedAssignedPartitions.iterator();
- TopicPartition nextAssignedPartition = sortedAssignedPartitionsIter.next();
-
- for (String topic : sortedAllTopics) {
- int partitionCount = partitionsPerTopic.get(topic);
- for (int i = 0; i < partitionCount; i++) {
- if (shouldAddDirectly || !(nextAssignedPartition.topic().equals(topic) && nextAssignedPartition.partition() == i)) {
- unassignedPartitions.add(new TopicPartition(topic, i));
- } else {
- // this partition is in assignedPartitions, don't add to unassignedPartitions, just get next assigned partition
- if (sortedAssignedPartitionsIter.hasNext()) {
- nextAssignedPartition = sortedAssignedPartitionsIter.next();
- } else {
- // add the remaining directly since there is no more sortedAssignedPartitions
- shouldAddDirectly = true;
- }
- }
- }
- }
-
- return unassignedPartitions;
- }
-
private List getAllTopicPartitions(Map partitionsPerTopic,
List sortedAllTopics,
@@ -384,37 +326,39 @@ private List getAllTopicPartitions(Map partitio
*
* @param partitionsPerTopic The number of partitions for each subscribed topic.
* @param subscriptions Map from the member id to their respective topic subscription
+ * @param currentAssignment Each consumer's previously owned and still-subscribed partitions
*
* @return Map from each member to the list of partitions assigned to them.
*/
private Map> generalAssign(Map partitionsPerTopic,
- Map subscriptions) {
- Map> currentAssignment = new HashMap<>();
+ Map subscriptions,
+ Map> currentAssignment) {
+ if (log.isDebugEnabled()) {
+ log.debug("performing general assign. partitionsPerTopic: {}, subscriptions: {}, currentAssignment: {}",
+ partitionsPerTopic, subscriptions, currentAssignment);
+ }
+
Map prevAssignment = new HashMap<>();
partitionMovements = new PartitionMovements();
- prepopulateCurrentAssignments(subscriptions, currentAssignment, prevAssignment);
+ prepopulateCurrentAssignments(subscriptions, prevAssignment);
- // a mapping of all topic partitions to all consumers that can be assigned to them
- final Map> partition2AllPotentialConsumers = new HashMap<>();
- // a mapping of all consumers to all potential topic partitions that can be assigned to them
- final Map> consumer2AllPotentialPartitions = new HashMap<>();
+ // a mapping of all topics to all consumers that can be assigned to them
+ final Map> topic2AllPotentialConsumers = new HashMap<>(partitionsPerTopic.keySet().size());
+ // a mapping of all consumers to all potential topics that can be assigned to them
+ final Map> consumer2AllPotentialTopics = new HashMap<>(subscriptions.keySet().size());
- // initialize partition2AllPotentialConsumers and consumer2AllPotentialPartitions in the following two for loops
- for (Entry entry: partitionsPerTopic.entrySet()) {
- for (int i = 0; i < entry.getValue(); ++i)
- partition2AllPotentialConsumers.put(new TopicPartition(entry.getKey(), i), new ArrayList<>());
- }
+ // initialize topic2AllPotentialConsumers and consumer2AllPotentialTopics
+ partitionsPerTopic.keySet().stream().forEach(
+ topicName -> topic2AllPotentialConsumers.put(topicName, new ArrayList<>()));
for (Entry entry: subscriptions.entrySet()) {
String consumerId = entry.getKey();
- consumer2AllPotentialPartitions.put(consumerId, new ArrayList<>());
+ List subscribedTopics = new ArrayList<>(entry.getValue().topics().size());
+ consumer2AllPotentialTopics.put(consumerId, subscribedTopics);
entry.getValue().topics().stream().filter(topic -> partitionsPerTopic.get(topic) != null).forEach(topic -> {
- for (int i = 0; i < partitionsPerTopic.get(topic); ++i) {
- TopicPartition topicPartition = new TopicPartition(topic, i);
- consumer2AllPotentialPartitions.get(consumerId).add(topicPartition);
- partition2AllPotentialConsumers.get(topicPartition).add(consumerId);
- }
+ subscribedTopics.add(topic);
+ topic2AllPotentialConsumers.get(topic).add(consumerId);
});
// add this consumer to currentAssignment (with an empty topic partition assignment) if it does not already exist
@@ -428,14 +372,18 @@ private Map> generalAssign(Map par
for (TopicPartition topicPartition: entry.getValue())
currentPartitionConsumer.put(topicPartition, entry.getKey());
- List sortedPartitions = sortPartitions(partition2AllPotentialConsumers);
+ int totalPartitionsCount = partitionsPerTopic.values().stream().reduce(0, Integer::sum);
+ List sortedAllTopics = new ArrayList<>(topic2AllPotentialConsumers.keySet());
+ Collections.sort(sortedAllTopics, new TopicComparator(topic2AllPotentialConsumers));
+ List sortedAllPartitions = getAllTopicPartitions(partitionsPerTopic, sortedAllTopics, totalPartitionsCount);
- // all partitions that need to be assigned (initially set to all partitions but adjusted in the following loop)
- List unassignedPartitions = new ArrayList<>(sortedPartitions);
+ // the partitions already assigned in current assignment
+ List assignedPartitions = new ArrayList<>();
boolean revocationRequired = false;
for (Iterator>> it = currentAssignment.entrySet().iterator(); it.hasNext();) {
Map.Entry> entry = it.next();
- if (!subscriptions.containsKey(entry.getKey())) {
+ Subscription consumerSubscription = subscriptions.get(entry.getKey());
+ if (consumerSubscription == null) {
// if a consumer that existed before (and had some partition assignments) is now removed, remove it from currentAssignment
for (TopicPartition topicPartition: entry.getValue())
currentPartitionConsumer.remove(topicPartition);
@@ -444,23 +392,32 @@ private Map> generalAssign(Map par
// otherwise (the consumer still exists)
for (Iterator partitionIter = entry.getValue().iterator(); partitionIter.hasNext();) {
TopicPartition partition = partitionIter.next();
- if (!partition2AllPotentialConsumers.containsKey(partition)) {
- // if this topic partition of this consumer no longer exists remove it from currentAssignment of the consumer
+ if (!topic2AllPotentialConsumers.containsKey(partition.topic())) {
+ // if this topic partition of this consumer no longer exists, remove it from currentAssignment of the consumer
partitionIter.remove();
currentPartitionConsumer.remove(partition);
- } else if (!subscriptions.get(entry.getKey()).topics().contains(partition.topic())) {
- // if this partition cannot remain assigned to its current consumer because the consumer
- // is no longer subscribed to its topic remove it from currentAssignment of the consumer
+ } else if (!consumerSubscription.topics().contains(partition.topic())) {
+ // because the consumer is no longer subscribed to its topic, remove it from currentAssignment of the consumer
partitionIter.remove();
revocationRequired = true;
- } else
+ } else {
// otherwise, remove the topic partition from those that need to be assigned only if
// its current consumer is still subscribed to its topic (because it is already assigned
// and we would want to preserve that assignment as much as possible)
- unassignedPartitions.remove(partition);
+ assignedPartitions.add(partition);
+ }
}
}
}
+
+ // all partitions that needed to be assigned
+ List unassignedPartitions = getUnassignedPartitions(sortedAllPartitions, assignedPartitions, topic2AllPotentialConsumers);
+ assignedPartitions = null;
+
+ if (log.isDebugEnabled()) {
+ log.debug("unassigned Partitions: {}", unassignedPartitions);
+ }
+
// at this point we have preserved all valid topic partition to consumer assignments and removed
// all invalid topic partitions and invalid consumers. Now we need to assign unassignedPartitions
// to consumers so that the topic partition assignments are as balanced as possible.
@@ -469,58 +426,172 @@ private Map> generalAssign(Map par
TreeSet sortedCurrentSubscriptions = new TreeSet<>(new SubscriptionComparator(currentAssignment));
sortedCurrentSubscriptions.addAll(currentAssignment.keySet());
- balance(currentAssignment, prevAssignment, sortedPartitions, unassignedPartitions, sortedCurrentSubscriptions,
- consumer2AllPotentialPartitions, partition2AllPotentialConsumers, currentPartitionConsumer, revocationRequired);
+ balance(currentAssignment, prevAssignment, sortedAllPartitions, unassignedPartitions, sortedCurrentSubscriptions,
+ consumer2AllPotentialTopics, topic2AllPotentialConsumers, currentPartitionConsumer, revocationRequired,
+ partitionsPerTopic, totalPartitionsCount);
+
+ if (log.isDebugEnabled()) {
+ log.debug("final assignment: {}", currentAssignment);
+ }
+
return currentAssignment;
}
+ /**
+ * get the unassigned partition list by computing the difference set of the sortedPartitions(all partitions)
+ * and sortedAssignedPartitions. If no assigned partitions, we'll just return all sorted topic partitions.
+ * This is used in generalAssign method
+ *
+ * We loop the sortedPartition, and compare the ith element in sortedAssignedPartitions(i start from 0):
+ * - if not equal to the ith element, add to unassignedPartitions
+ * - if equal to the the ith element, get next element from sortedAssignedPartitions
+ *
+ * @param sortedAllPartitions: sorted all partitions
+ * @param sortedAssignedPartitions: sorted partitions, all are included in the sortedPartitions
+ * @param topic2AllPotentialConsumers: topics mapped to all consumers that subscribed to it
+ * @return partitions that aren't assigned to any current consumer
+ */
+ private List getUnassignedPartitions(List sortedAllPartitions,
+ List sortedAssignedPartitions,
+ Map> topic2AllPotentialConsumers) {
+ if (sortedAssignedPartitions.isEmpty()) {
+ return sortedAllPartitions;
+ }
+
+ List unassignedPartitions = new ArrayList<>();
+
+ Collections.sort(sortedAssignedPartitions, new PartitionComparator(topic2AllPotentialConsumers));
+
+ boolean shouldAddDirectly = false;
+ Iterator sortedAssignedPartitionsIter = sortedAssignedPartitions.iterator();
+ TopicPartition nextAssignedPartition = sortedAssignedPartitionsIter.next();
+
+ for (TopicPartition topicPartition : sortedAllPartitions) {
+ if (shouldAddDirectly || !nextAssignedPartition.equals(topicPartition)) {
+ unassignedPartitions.add(topicPartition);
+ } else {
+ // this partition is in assignedPartitions, don't add to unassignedPartitions, just get next assigned partition
+ if (sortedAssignedPartitionsIter.hasNext()) {
+ nextAssignedPartition = sortedAssignedPartitionsIter.next();
+ } else {
+ // add the remaining directly since there is no more sortedAssignedPartitions
+ shouldAddDirectly = true;
+ }
+ }
+ }
+ return unassignedPartitions;
+ }
+
+ /**
+ * get the unassigned partition list by computing the difference set of all sorted partitions
+ * and sortedAssignedPartitions. If no assigned partitions, we'll just return all sorted topic partitions.
+ * This is used in constrainedAssign method
+ *
+ * To compute the difference set, we use two pointers technique here:
+ *
+ * We loop through the all sorted topics, and then iterate all partitions the topic has,
+ * compared with the ith element in sortedAssignedPartitions(i starts from 0):
+ * - if not equal to the ith element, add to unassignedPartitions
+ * - if equal to the the ith element, get next element from sortedAssignedPartitions
+ *
+ * @param totalPartitionsCount all partitions counts in this assignment
+ * @param partitionsPerTopic the number of partitions for each subscribed topic.
+ * @param sortedAssignedPartitions sorted partitions, all are included in the sortedPartitions
+ * @return the partitions not yet assigned to any consumers
+ */
+ private List