diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AbortTransactionOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AbortTransactionOptions.java index ad5827ac6d0f5..52dc6b10bb964 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AbortTransactionOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AbortTransactionOptions.java @@ -20,4 +20,12 @@ @InterfaceStability.Evolving public class AbortTransactionOptions extends AbstractOptions { + + @Override + public String toString() { + return "AbortTransactionOptions(" + + "timeoutMs=" + timeoutMs + + ')'; + } + } 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 3e52f64064af0..3dd2acc33c0d0 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 @@ -1533,6 +1533,28 @@ default AbortTransactionResult abortTransaction(AbortTransactionSpec spec) { */ AbortTransactionResult abortTransaction(AbortTransactionSpec spec, AbortTransactionOptions options); + /** + * List active transactions in the cluster. See + * {@link #listTransactions(ListTransactionsOptions)} for more details. + * + * @return The result + */ + default ListTransactionsResult listTransactions() { + return listTransactions(new ListTransactionsOptions()); + } + + /** + * List active transactions in the cluster. This will query all potential transaction + * coordinators in the cluster and collect the state of all transactions. Users + * should typically attempt to reduce the size of the result set using + * {@link ListTransactionsOptions#filterProducerIds(Collection)} or + * {@link ListTransactionsOptions#filterStates(Collection)} + * + * @param options Options to control the method behavior (including filters) + * @return The result + */ + ListTransactionsResult listTransactions(ListTransactionsOptions options); + /** * Get the metrics kept by the adminClient */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeProducersResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeProducersResult.java index 473191661f349..55897bfc01c7c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeProducersResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeProducersResult.java @@ -71,6 +71,13 @@ public PartitionProducerState(List activeProducers) { public List activeProducers() { return activeProducers; } + + @Override + public String toString() { + return "PartitionProducerState(" + + "activeProducers=" + activeProducers + + ')'; + } } } diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTransactionsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTransactionsOptions.java index 3759b49f8abca..47beb7f343c62 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTransactionsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTransactionsOptions.java @@ -28,4 +28,11 @@ @InterfaceStability.Evolving public class DescribeTransactionsOptions extends AbstractOptions { + @Override + public String toString() { + return "DescribeTransactionsOptions(" + + "timeoutMs=" + timeoutMs + + ')'; + } + } 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 e49f28a62f61b..6477bd060434e 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 @@ -34,10 +34,14 @@ import org.apache.kafka.clients.admin.internals.AbortTransactionHandler; import org.apache.kafka.clients.admin.internals.AdminApiDriver; import org.apache.kafka.clients.admin.internals.AdminApiHandler; +import org.apache.kafka.clients.admin.internals.AdminApiFuture; import org.apache.kafka.clients.admin.internals.AdminMetadataManager; +import org.apache.kafka.clients.admin.internals.AllBrokersStrategy; import org.apache.kafka.clients.admin.internals.ConsumerGroupOperationContext; +import org.apache.kafka.clients.admin.internals.CoordinatorKey; import org.apache.kafka.clients.admin.internals.DescribeProducersHandler; import org.apache.kafka.clients.admin.internals.DescribeTransactionsHandler; +import org.apache.kafka.clients.admin.internals.ListTransactionsHandler; import org.apache.kafka.clients.admin.internals.MetadataOperationContext; import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Assignment; import org.apache.kafka.clients.consumer.OffsetAndMetadata; @@ -4729,34 +4733,43 @@ void handleFailure(Throwable throwable) { @Override public DescribeProducersResult describeProducers(Collection topicPartitions, DescribeProducersOptions options) { - DescribeProducersHandler handler = new DescribeProducersHandler( - new HashSet<>(topicPartitions), - options, - logContext - ); - return new DescribeProducersResult(invokeDriver(handler, options.timeoutMs)); + AdminApiFuture.SimpleAdminApiFuture future = + DescribeProducersHandler.newFuture(topicPartitions); + DescribeProducersHandler handler = new DescribeProducersHandler(options, logContext); + invokeDriver(handler, future, options.timeoutMs); + return new DescribeProducersResult(future.all()); } @Override public DescribeTransactionsResult describeTransactions(Collection transactionalIds, DescribeTransactionsOptions options) { - DescribeTransactionsHandler handler = new DescribeTransactionsHandler( - transactionalIds, - logContext - ); - return new DescribeTransactionsResult(invokeDriver(handler, options.timeoutMs)); + AdminApiFuture.SimpleAdminApiFuture future = + DescribeTransactionsHandler.newFuture(transactionalIds); + DescribeTransactionsHandler handler = new DescribeTransactionsHandler(logContext); + invokeDriver(handler, future, options.timeoutMs); + return new DescribeTransactionsResult(future.all()); } @Override public AbortTransactionResult abortTransaction(AbortTransactionSpec spec, AbortTransactionOptions options) { - AbortTransactionHandler handler = new AbortTransactionHandler( - spec, - logContext - ); - return new AbortTransactionResult(invokeDriver(handler, options.timeoutMs)); + AdminApiFuture.SimpleAdminApiFuture future = + AbortTransactionHandler.newFuture(Collections.singleton(spec.topicPartition())); + AbortTransactionHandler handler = new AbortTransactionHandler(spec, logContext); + invokeDriver(handler, future, options.timeoutMs); + return new AbortTransactionResult(future.all()); + } + + @Override + public ListTransactionsResult listTransactions(ListTransactionsOptions options) { + AllBrokersStrategy.AllBrokersFuture> future = + ListTransactionsHandler.newFuture(); + ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); + invokeDriver(handler, future, options.timeoutMs); + return new ListTransactionsResult(future.all()); } - private Map> invokeDriver( + private void invokeDriver( AdminApiHandler handler, + AdminApiFuture future, Integer timeoutMs ) { long currentTimeMs = time.milliseconds(); @@ -4764,13 +4777,13 @@ private Map> invokeDriver( AdminApiDriver driver = new AdminApiDriver<>( handler, + future, deadlineMs, retryBackoffMs, logContext ); maybeSendRequests(driver, currentTimeMs); - return driver.futures(); } private void maybeSendRequests(AdminApiDriver driver, long currentTimeMs) { diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java new file mode 100644 index 0000000000000..3b2d90284d12b --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListTransactionsOptions.java @@ -0,0 +1,91 @@ +/* + * 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.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +/** + * Options for {@link Admin#listTransactions()}. + * + * The API of this class is evolving, see {@link Admin} for details. + */ +@InterfaceStability.Evolving +public class ListTransactionsOptions extends AbstractOptions { + private Set filteredStates = Collections.emptySet(); + private Set filteredProducerIds = Collections.emptySet(); + + /** + * Filter only the transactions that are in a specific set of states. If no filter + * is specified or if the passed set of states is empty, then transactions in all + * states will be returned. + * + * @param states the set of states to filter by + * @return this object + */ + public ListTransactionsOptions filterStates(Collection states) { + this.filteredStates = new HashSet<>(states); + return this; + } + + /** + * Filter only the transactions from producers in a specific set of producerIds. + * If no filter is specified or if the passed collection of producerIds is empty, + * then the transactions of all producerIds will be returned. + * + * @param producerIdFilters the set of producerIds to filter by + * @return this object + */ + public ListTransactionsOptions filterProducerIds(Collection producerIdFilters) { + this.filteredProducerIds = new HashSet<>(producerIdFilters); + return this; + } + + /** + * Returns the set of states to be filtered or empty if no states have been specified. + * + * @return the current set of filtered states (empty means that no states are filtered and all + * all transactions will be returned) + */ + public Set filteredStates() { + return filteredStates; + } + + /** + * Returns the set of producerIds that are being filtered or empty if none have been specified. + * + * @return the current set of filtered states (empty means that no producerIds are filtered and + * all transactions will be returned) + */ + public Set 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/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java index 8c987a12d02fa..0013d33d196ae 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java @@ -115,6 +115,7 @@ import org.apache.kafka.common.message.ListOffsetsResponseData; import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsTopicResponse; import org.apache.kafka.common.message.ListPartitionReassignmentsResponseData; +import org.apache.kafka.common.message.ListTransactionsResponseData; import org.apache.kafka.common.message.MetadataResponseData; import org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition; import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic; @@ -173,6 +174,8 @@ import org.apache.kafka.common.requests.ListGroupsResponse; import org.apache.kafka.common.requests.ListOffsetsResponse; import org.apache.kafka.common.requests.ListPartitionReassignmentsResponse; +import org.apache.kafka.common.requests.ListTransactionsRequest; +import org.apache.kafka.common.requests.ListTransactionsResponse; import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.requests.OffsetCommitResponse; @@ -5621,6 +5624,58 @@ public void testAbortTransactionFindLeaderAfterDisconnect() throws Exception { } } + @Test + public void testListTransactions() throws Exception { + try (AdminClientUnitTestEnv env = mockClientEnv()) { + MetadataResponseData.MetadataResponseBrokerCollection brokers = + new MetadataResponseData.MetadataResponseBrokerCollection(); + + env.cluster().nodes().forEach(node -> { + brokers.add(new MetadataResponseData.MetadataResponseBroker() + .setHost(node.host()) + .setNodeId(node.id()) + .setPort(node.port()) + .setRack(node.rack()) + ); + }); + + env.kafkaClient().prepareResponse( + request -> request instanceof MetadataRequest, + new MetadataResponse(new MetadataResponseData().setBrokers(brokers), + MetadataResponseData.HIGHEST_SUPPORTED_VERSION) + ); + + List expected = Arrays.asList( + new TransactionListing("foo", 12345L, TransactionState.ONGOING), + new TransactionListing("bar", 98765L, TransactionState.PREPARE_ABORT), + new TransactionListing("baz", 13579L, TransactionState.COMPLETE_COMMIT) + ); + assertEquals(Utils.mkSet(0, 1, 2), env.cluster().nodes().stream().map(Node::id) + .collect(Collectors.toSet())); + + env.cluster().nodes().forEach(node -> { + ListTransactionsResponseData response = new ListTransactionsResponseData() + .setErrorCode(Errors.NONE.code()); + + TransactionListing listing = expected.get(node.id()); + response.transactionStates().add(new ListTransactionsResponseData.TransactionState() + .setTransactionalId(listing.transactionalId()) + .setProducerId(listing.producerId()) + .setTransactionState(listing.state().toString()) + ); + + env.kafkaClient().prepareResponseFrom( + request -> request instanceof ListTransactionsRequest, + new ListTransactionsResponse(response), + node + ); + }); + + ListTransactionsResult result = env.adminClient().listTransactions(); + assertEquals(new HashSet<>(expected), new HashSet<>(result.all().get())); + } + } + private WriteTxnMarkersResponse writeTxnMarkersResponse( AbortTransactionSpec abortSpec, Errors error diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/ListTransactionsResultTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/ListTransactionsResultTest.java new file mode 100644 index 0000000000000..769c50b8518c1 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/admin/ListTransactionsResultTest.java @@ -0,0 +1,119 @@ +/* + * 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.KafkaException; +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.internals.KafkaFutureImpl; +import org.apache.kafka.common.utils.Utils; +import org.junit.jupiter.api.Test; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static org.apache.kafka.test.TestUtils.assertFutureThrows; +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class ListTransactionsResultTest { + private final KafkaFutureImpl>>> future = + new KafkaFutureImpl<>(); + private final ListTransactionsResult result = new ListTransactionsResult(future); + + @Test + public void testAllFuturesFailIfLookupFails() { + future.completeExceptionally(new KafkaException()); + assertFutureThrows(result.all(), KafkaException.class); + assertFutureThrows(result.allByBrokerId(), KafkaException.class); + assertFutureThrows(result.byBrokerId(), KafkaException.class); + } + + @Test + public void testAllFuturesSucceed() throws Exception { + KafkaFutureImpl> future1 = new KafkaFutureImpl<>(); + KafkaFutureImpl> future2 = new KafkaFutureImpl<>(); + + Map>> brokerFutures = new HashMap<>(); + brokerFutures.put(1, future1); + brokerFutures.put(2, future2); + + future.complete(brokerFutures); + + List broker1Listings = asList( + new TransactionListing("foo", 12345L, TransactionState.ONGOING), + new TransactionListing("bar", 98765L, TransactionState.PREPARE_ABORT) + ); + future1.complete(broker1Listings); + + List broker2Listings = singletonList( + new TransactionListing("baz", 13579L, TransactionState.COMPLETE_COMMIT) + ); + future2.complete(broker2Listings); + + Map>> resultBrokerFutures = + result.byBrokerId().get(); + + assertEquals(Utils.mkSet(1, 2), resultBrokerFutures.keySet()); + assertEquals(broker1Listings, resultBrokerFutures.get(1).get()); + assertEquals(broker2Listings, resultBrokerFutures.get(2).get()); + assertEquals(broker1Listings, result.allByBrokerId().get().get(1)); + assertEquals(broker2Listings, result.allByBrokerId().get().get(2)); + + Set allExpected = new HashSet<>(); + allExpected.addAll(broker1Listings); + allExpected.addAll(broker2Listings); + + assertEquals(allExpected, new HashSet<>(result.all().get())); + } + + @Test + public void testPartialFailure() throws Exception { + KafkaFutureImpl> future1 = new KafkaFutureImpl<>(); + KafkaFutureImpl> future2 = new KafkaFutureImpl<>(); + + Map>> brokerFutures = new HashMap<>(); + brokerFutures.put(1, future1); + brokerFutures.put(2, future2); + + future.complete(brokerFutures); + + List broker1Listings = asList( + new TransactionListing("foo", 12345L, TransactionState.ONGOING), + new TransactionListing("bar", 98765L, TransactionState.PREPARE_ABORT) + ); + future1.complete(broker1Listings); + future2.completeExceptionally(new KafkaException()); + + Map>> resultBrokerFutures = + result.byBrokerId().get(); + + // Ensure that the future for broker 1 completes successfully + assertEquals(Utils.mkSet(1, 2), resultBrokerFutures.keySet()); + assertEquals(broker1Listings, resultBrokerFutures.get(1).get()); + + // Everything else should fail + assertFutureThrows(result.all(), KafkaException.class); + assertFutureThrows(result.allByBrokerId(), KafkaException.class); + assertFutureThrows(resultBrokerFutures.get(2), KafkaException.class); + } + +} diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java index 369e1e6f55fe8..73e065f373a18 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java @@ -918,6 +918,11 @@ public AbortTransactionResult abortTransaction(AbortTransactionSpec spec, AbortT throw new UnsupportedOperationException("Not implemented yet"); } + @Override + public ListTransactionsResult listTransactions(ListTransactionsOptions options) { + throw new UnsupportedOperationException("Not implemented yet"); + } + @Override synchronized public void close(Duration timeout) {} diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java index 06e72dfdb8182..a78b73a0fb645 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java @@ -18,7 +18,6 @@ import org.apache.kafka.clients.admin.internals.AdminApiDriver.RequestSpec; import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult; -import org.apache.kafka.clients.admin.internals.AdminApiHandler.Keys; import org.apache.kafka.clients.admin.internals.AdminApiLookupStrategy.LookupResult; import org.apache.kafka.common.errors.DisconnectException; import org.apache.kafka.common.errors.UnknownServerException; @@ -38,6 +37,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.OptionalInt; import java.util.Set; @@ -57,11 +57,10 @@ class AdminApiDriverTest { @Test public void testCoalescedLookup() { - MockRequestScope scope = new MockRequestScope(OptionalInt.empty()); - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", scope, - "bar", scope - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1", + "bar", "c1" + )); Map, LookupResult> lookupRequests = map( mkSet("foo", "bar"), mapped("foo", 1, "bar", 2) @@ -81,10 +80,10 @@ public void testCoalescedLookup() { @Test public void testCoalescedFulfillment() { - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", new MockRequestScope(OptionalInt.empty()), - "bar", new MockRequestScope(OptionalInt.of(1)) - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1", + "bar", "c2" + )); Map, LookupResult> lookupRequests = map( mkSet("foo"), mapped("foo", 1), @@ -104,10 +103,10 @@ public void testCoalescedFulfillment() { @Test public void testKeyLookupFailure() { - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", new MockRequestScope(OptionalInt.empty()), - "bar", new MockRequestScope(OptionalInt.of(1)) - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1", + "bar", "c2" + )); Map, LookupResult> lookupRequests = map( mkSet("foo"), failedLookup("foo", new UnknownServerException()), @@ -127,10 +126,10 @@ public void testKeyLookupFailure() { @Test public void testKeyLookupRetry() { - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", new MockRequestScope(OptionalInt.empty()), - "bar", new MockRequestScope(OptionalInt.of(1)) - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1", + "bar", "c2" + )); Map, LookupResult> lookupRequests = map( mkSet("foo"), emptyLookup(), @@ -160,11 +159,11 @@ public void testKeyLookupRetry() { @Test public void testStaticMapping() { - TestContext ctx = new TestContext(Keys.staticMapped(map( + TestContext ctx = TestContext.staticMapped(map( "foo", 0, "bar", 1, "baz", 1 - ))); + )); Map, ApiResult> fulfillmentResults = map( mkSet("foo"), completed("foo", 15L), @@ -178,11 +177,11 @@ public void testStaticMapping() { @Test public void testFulfillmentFailure() { - TestContext ctx = new TestContext(Keys.staticMapped(map( + TestContext ctx = TestContext.staticMapped(map( "foo", 0, "bar", 1, "baz", 1 - ))); + )); Map, ApiResult> fulfillmentResults = map( mkSet("foo"), failed("foo", new UnknownServerException()), @@ -196,11 +195,11 @@ public void testFulfillmentFailure() { @Test public void testFulfillmentRetry() { - TestContext ctx = new TestContext(Keys.staticMapped(map( + TestContext ctx = TestContext.staticMapped(map( "foo", 0, "bar", 1, "baz", 1 - ))); + )); Map, ApiResult> fulfillmentResults = map( mkSet("foo"), completed("foo", 15L), @@ -220,10 +219,10 @@ public void testFulfillmentRetry() { @Test public void testFulfillmentUnmapping() { - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", new MockRequestScope(OptionalInt.empty()), - "bar", new MockRequestScope(OptionalInt.of(1)) - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1", + "bar", "c2" + )); Map, LookupResult> lookupRequests = map( mkSet("foo"), mapped("foo", 0), @@ -256,11 +255,10 @@ public void testFulfillmentUnmapping() { @Test public void testRecoalescedLookup() { - MockRequestScope scope = new MockRequestScope(OptionalInt.empty()); - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", scope, - "bar", scope - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1", + "bar", "c1" + )); Map, LookupResult> lookupRequests = map( mkSet("foo", "bar"), mapped("foo", 1, "bar", 2) @@ -292,9 +290,9 @@ public void testRecoalescedLookup() { @Test public void testRetryLookupAfterDisconnect() { - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", new MockRequestScope(OptionalInt.empty()) - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1" + )); int initialLeaderId = 1; @@ -303,7 +301,7 @@ public void testRetryLookupAfterDisconnect() { ); ctx.poll(initialLookup, emptyMap()); - assertMappedKey(ctx.driver, "foo", initialLeaderId); + assertMappedKey(ctx, "foo", initialLeaderId); ctx.handler.expectRequest(mkSet("foo"), completed("foo", 15L)); @@ -314,7 +312,7 @@ public void testRetryLookupAfterDisconnect() { assertEquals(OptionalInt.of(initialLeaderId), requestSpec.scope.destinationBrokerId()); ctx.driver.onFailure(ctx.time.milliseconds(), requestSpec, new DisconnectException()); - assertUnmappedKey(ctx.driver, "foo"); + assertUnmappedKey(ctx, "foo"); int retryLeaderId = 2; @@ -329,19 +327,18 @@ public void testRetryLookupAfterDisconnect() { @Test public void testCoalescedStaticAndDynamicFulfillment() { - Map dynamicLookupScopes = map( - "foo", new MockRequestScope(OptionalInt.empty()) + Map dynamicMapping = map( + "foo", "c1" ); Map staticMapping = map( "bar", 1 ); - TestContext ctx = new TestContext(new Keys<>( + TestContext ctx = new TestContext( staticMapping, - dynamicLookupScopes.keySet(), - new MockLookupStrategy<>(dynamicLookupScopes) - )); + dynamicMapping + ); // Initially we expect a lookup for the dynamic key and a // fulfillment request for the static key @@ -400,9 +397,9 @@ public void testCoalescedStaticAndDynamicFulfillment() { @Test public void testLookupRetryBookkeeping() { - TestContext ctx = new TestContext(dynamicMapped(map( - "foo", new MockRequestScope(OptionalInt.empty()) - ))); + TestContext ctx = TestContext.dynamicMapped(map( + "foo", "c1" + )); LookupResult emptyLookup = emptyLookup(); ctx.lookupStrategy().expectLookup(mkSet("foo"), emptyLookup); @@ -425,7 +422,7 @@ public void testLookupRetryBookkeeping() { @Test public void testFulfillmentRetryBookkeeping() { - TestContext ctx = new TestContext(Keys.staticMapped(map("foo", 0))); + TestContext ctx = TestContext.staticMapped(map("foo", 0)); ApiResult emptyFulfillment = emptyFulfillment(); ctx.handler.expectRequest(mkSet("foo"), emptyFulfillment); @@ -447,41 +444,41 @@ public void testFulfillmentRetryBookkeeping() { } private static void assertMappedKey( - AdminApiDriver driver, + TestContext context, String key, Integer expectedBrokerId ) { - OptionalInt brokerIdOpt = driver.keyToBrokerId(key); + OptionalInt brokerIdOpt = context.driver.keyToBrokerId(key); assertEquals(OptionalInt.of(expectedBrokerId), brokerIdOpt); } private static void assertUnmappedKey( - AdminApiDriver driver, + TestContext context, String key ) { - OptionalInt brokerIdOpt = driver.keyToBrokerId(key); + OptionalInt brokerIdOpt = context.driver.keyToBrokerId(key); assertEquals(OptionalInt.empty(), brokerIdOpt); - KafkaFutureImpl future = driver.futures().get(key); + KafkaFutureImpl future = context.future.all().get(key); assertFalse(future.isDone()); } private static void assertFailedKey( - AdminApiDriver driver, + TestContext context, String key, Throwable expectedException ) { - KafkaFutureImpl future = driver.futures().get(key); + KafkaFutureImpl future = context.future.all().get(key); assertTrue(future.isCompletedExceptionally()); Throwable exception = assertThrows(ExecutionException.class, future::get); assertEquals(expectedException, exception.getCause()); } private static void assertCompletedKey( - AdminApiDriver driver, + TestContext context, String key, Long expected ) { - KafkaFutureImpl future = driver.futures().get(key); + KafkaFutureImpl future = context.future.all().get(key); assertTrue(future.isDone()); try { assertEquals(expected, future.get()); @@ -492,48 +489,92 @@ private static void assertCompletedKey( private static class MockRequestScope implements ApiRequestScope { private final OptionalInt destinationBrokerId; + private final String id; - private MockRequestScope(OptionalInt destinationBrokerId) { + private MockRequestScope( + OptionalInt destinationBrokerId, + String id + ) { this.destinationBrokerId = destinationBrokerId; + this.id = id; } @Override public OptionalInt destinationBrokerId() { return destinationBrokerId; } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + MockRequestScope that = (MockRequestScope) o; + return Objects.equals(destinationBrokerId, that.destinationBrokerId) && + Objects.equals(id, that.id); + } + + @Override + public int hashCode() { + return Objects.hash(destinationBrokerId, id); + } } private static class TestContext { private final MockTime time = new MockTime(); - private final Keys keys; private final MockAdminApiHandler handler; private final AdminApiDriver driver; + private final AdminApiFuture.SimpleAdminApiFuture future; + + public TestContext( + Map staticKeys, + Map dynamicKeys + ) { + Map lookupScopes = new HashMap<>(); + staticKeys.forEach((key, brokerId) -> { + MockRequestScope scope = new MockRequestScope(OptionalInt.of(brokerId), null); + lookupScopes.put(key, scope); + }); + + dynamicKeys.forEach((key, context) -> { + MockRequestScope scope = new MockRequestScope(OptionalInt.empty(), context); + lookupScopes.put(key, scope); + }); + + MockLookupStrategy lookupStrategy = new MockLookupStrategy<>(lookupScopes); + this.handler = new MockAdminApiHandler<>(lookupStrategy); + this.future = AdminApiFuture.forKeys(lookupStrategy.lookupScopes.keySet()); - public TestContext(Keys keys) { - this.keys = keys; - this.handler = new MockAdminApiHandler<>(keys); this.driver = new AdminApiDriver<>( handler, + future, time.milliseconds() + API_TIMEOUT_MS, RETRY_BACKOFF_MS, new LogContext() ); - keys.staticKeys.forEach((key, brokerId) -> { - assertMappedKey(driver, key, brokerId); + staticKeys.forEach((key, brokerId) -> { + assertMappedKey(this, key, brokerId); }); - keys.dynamicKeys.forEach(key -> { - assertUnmappedKey(driver, key); + dynamicKeys.keySet().forEach(key -> { + assertUnmappedKey(this, key); }); } + public static TestContext staticMapped(Map staticKeys) { + return new TestContext(staticKeys, Collections.emptyMap()); + } + + public static TestContext dynamicMapped(Map dynamicKeys) { + return new TestContext(Collections.emptyMap(), dynamicKeys); + } + private void assertLookupResponse( RequestSpec requestSpec, LookupResult result ) { requestSpec.keys.forEach(key -> { - assertUnmappedKey(driver, key); + assertUnmappedKey(this, key); }); // The response is just a placeholder. The result is all we are interested in @@ -542,11 +583,11 @@ private void assertLookupResponse( driver.onResponse(time.milliseconds(), requestSpec, response); result.mappedKeys.forEach((key, brokerId) -> { - assertMappedKey(driver, key, brokerId); + assertMappedKey(this, key, brokerId); }); result.failedKeys.forEach((key, exception) -> { - assertFailedKey(driver, key, exception); + assertFailedKey(this, key, exception); }); } @@ -558,7 +599,7 @@ private void assertResponse( new AssertionError("Fulfillment requests must specify a target brokerId")); requestSpec.keys.forEach(key -> { - assertMappedKey(driver, key, brokerId); + assertMappedKey(this, key, brokerId); }); // The response is just a placeholder. The result is all we are interested in @@ -568,23 +609,20 @@ private void assertResponse( driver.onResponse(time.milliseconds(), requestSpec, response); result.unmappedKeys.forEach(key -> { - assertUnmappedKey(driver, key); + assertUnmappedKey(this, key); }); result.failedKeys.forEach((key, exception) -> { - assertFailedKey(driver, key, exception); + assertFailedKey(this, key, exception); }); result.completedKeys.forEach((key, value) -> { - assertCompletedKey(driver, key, value); + assertCompletedKey(this, key, value); }); } private MockLookupStrategy lookupStrategy() { - if (keys.dynamicKeys.isEmpty()) { - throw new IllegalStateException("Unexpected lookup when no dynamic mapping is defined"); - } - return (MockLookupStrategy) keys.lookupStrategy; + return handler.lookupStrategy; } public void poll( @@ -656,11 +694,11 @@ public void reset() { } private static class MockAdminApiHandler implements AdminApiHandler { - private final Keys keyMappings; private final Map, ApiResult> expectedRequests = new HashMap<>(); + private final MockLookupStrategy lookupStrategy; - private MockAdminApiHandler(Keys keyMappings) { - this.keyMappings = keyMappings; + private MockAdminApiHandler(MockLookupStrategy lookupStrategy) { + this.lookupStrategy = lookupStrategy; } @Override @@ -669,8 +707,8 @@ public String apiName() { } @Override - public Keys initializeKeys() { - return keyMappings; + public AdminApiLookupStrategy lookupStrategy() { + return lookupStrategy; } public void expectRequest(Set keys, ApiResult result) { @@ -715,11 +753,6 @@ private static Map map(K k1, V v1, K k2, V v2, K k3, V v3) { return map; } - private static Keys dynamicMapped(Map lookupScopes) { - MockLookupStrategy strategy = new MockLookupStrategy<>(lookupScopes); - return Keys.dynamicMapped(lookupScopes.keySet(), strategy); - } - private static ApiResult completed(String key, Long value) { return new ApiResult<>(map(key, value), emptyMap(), Collections.emptyList()); } diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java new file mode 100644 index 0000000000000..f4ec4680026e6 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java @@ -0,0 +1,247 @@ +/* + * 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.errors.DisconnectException; +import org.apache.kafka.common.errors.UnknownServerException; +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.protocol.ApiKeys; +import org.apache.kafka.common.requests.AbstractRequest; +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.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.test.TestUtils; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.OptionalInt; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class AllBrokersStrategyIntegrationTest { + private static final long TIMEOUT_MS = 5000; + private static final long RETRY_BACKOFF_MS = 100; + + private final LogContext logContext = new LogContext(); + private final MockTime time = new MockTime(); + + private AdminApiDriver buildDriver( + AllBrokersStrategy.AllBrokersFuture result + ) { + return new AdminApiDriver<>( + new MockApiHandler(), + result, + time.milliseconds() + TIMEOUT_MS, + RETRY_BACKOFF_MS, + logContext + ); + } + + @Test + public void testFatalLookupError() { + AllBrokersStrategy.AllBrokersFuture result = new AllBrokersStrategy.AllBrokersFuture<>(); + AdminApiDriver driver = buildDriver(result); + + List> requestSpecs = driver.poll(); + assertEquals(1, requestSpecs.size()); + + AdminApiDriver.RequestSpec spec = requestSpecs.get(0); + assertEquals(AllBrokersStrategy.LOOKUP_KEYS, spec.keys); + + driver.onFailure(time.milliseconds(), spec, new UnknownServerException()); + assertTrue(result.all().isDone()); + TestUtils.assertFutureThrows(result.all(), UnknownServerException.class); + assertEquals(Collections.emptyList(), driver.poll()); + } + + @Test + public void testRetryLookupAfterDisconnect() { + AllBrokersStrategy.AllBrokersFuture result = new AllBrokersStrategy.AllBrokersFuture<>(); + AdminApiDriver driver = buildDriver(result); + + List> requestSpecs = driver.poll(); + assertEquals(1, requestSpecs.size()); + + AdminApiDriver.RequestSpec spec = requestSpecs.get(0); + assertEquals(AllBrokersStrategy.LOOKUP_KEYS, spec.keys); + + driver.onFailure(time.milliseconds(), spec, new DisconnectException()); + List> retrySpecs = driver.poll(); + assertEquals(1, retrySpecs.size()); + + AdminApiDriver.RequestSpec retrySpec = retrySpecs.get(0); + assertEquals(AllBrokersStrategy.LOOKUP_KEYS, retrySpec.keys); + assertEquals(time.milliseconds() + RETRY_BACKOFF_MS, retrySpec.nextAllowedTryMs); + assertEquals(Collections.emptyList(), driver.poll()); + } + + @Test + public void testMultiBrokerCompletion() throws Exception { + AllBrokersStrategy.AllBrokersFuture result = new AllBrokersStrategy.AllBrokersFuture<>(); + AdminApiDriver driver = buildDriver(result); + + List> lookupSpecs = driver.poll(); + assertEquals(1, lookupSpecs.size()); + AdminApiDriver.RequestSpec lookupSpec = lookupSpecs.get(0); + + Set brokerIds = Utils.mkSet(1, 2); + driver.onResponse(time.milliseconds(), lookupSpec, responseWithBrokers(brokerIds)); + assertTrue(result.all().isDone()); + + Map> brokerFutures = result.all().get(); + + List> requestSpecs = driver.poll(); + assertEquals(2, requestSpecs.size()); + + AdminApiDriver.RequestSpec requestSpec1 = requestSpecs.get(0); + assertTrue(requestSpec1.scope.destinationBrokerId().isPresent()); + int brokerId1 = requestSpec1.scope.destinationBrokerId().getAsInt(); + assertTrue(brokerIds.contains(brokerId1)); + + driver.onResponse(time.milliseconds(), requestSpec1, null); + KafkaFutureImpl future1 = brokerFutures.get(brokerId1); + assertTrue(future1.isDone()); + + AdminApiDriver.RequestSpec requestSpec2 = requestSpecs.get(1); + assertTrue(requestSpec2.scope.destinationBrokerId().isPresent()); + int brokerId2 = requestSpec2.scope.destinationBrokerId().getAsInt(); + assertNotEquals(brokerId1, brokerId2); + assertTrue(brokerIds.contains(brokerId2)); + + driver.onResponse(time.milliseconds(), requestSpec2, null); + KafkaFutureImpl future2 = brokerFutures.get(brokerId2); + assertTrue(future2.isDone()); + assertEquals(Collections.emptyList(), driver.poll()); + } + + @Test + public void testRetryFulfillmentAfterDisconnect() throws Exception { + AllBrokersStrategy.AllBrokersFuture result = new AllBrokersStrategy.AllBrokersFuture<>(); + AdminApiDriver driver = buildDriver(result); + + List> lookupSpecs = driver.poll(); + assertEquals(1, lookupSpecs.size()); + AdminApiDriver.RequestSpec lookupSpec = lookupSpecs.get(0); + + int brokerId = 1; + driver.onResponse(time.milliseconds(), lookupSpec, responseWithBrokers(Collections.singleton(brokerId))); + assertTrue(result.all().isDone()); + + Map> brokerFutures = result.all().get(); + KafkaFutureImpl future = brokerFutures.get(brokerId); + assertFalse(future.isDone()); + + List> requestSpecs = driver.poll(); + assertEquals(1, requestSpecs.size()); + AdminApiDriver.RequestSpec requestSpec = requestSpecs.get(0); + + driver.onFailure(time.milliseconds(), requestSpec, new DisconnectException()); + assertFalse(future.isDone()); + List> retrySpecs = driver.poll(); + assertEquals(1, retrySpecs.size()); + + AdminApiDriver.RequestSpec retrySpec = retrySpecs.get(0); + assertEquals(time.milliseconds() + RETRY_BACKOFF_MS, retrySpec.nextAllowedTryMs); + assertEquals(OptionalInt.of(brokerId), retrySpec.scope.destinationBrokerId()); + + driver.onResponse(time.milliseconds(), retrySpec, null); + assertTrue(future.isDone()); + assertEquals(brokerId, future.get()); + assertEquals(Collections.emptyList(), driver.poll()); + } + + @Test + public void testFatalFulfillmentError() throws Exception { + AllBrokersStrategy.AllBrokersFuture result = new AllBrokersStrategy.AllBrokersFuture<>(); + AdminApiDriver driver = buildDriver(result); + + List> lookupSpecs = driver.poll(); + assertEquals(1, lookupSpecs.size()); + AdminApiDriver.RequestSpec lookupSpec = lookupSpecs.get(0); + + int brokerId = 1; + driver.onResponse(time.milliseconds(), lookupSpec, responseWithBrokers(Collections.singleton(brokerId))); + assertTrue(result.all().isDone()); + + Map> brokerFutures = result.all().get(); + KafkaFutureImpl future = brokerFutures.get(brokerId); + assertFalse(future.isDone()); + + List> requestSpecs = driver.poll(); + assertEquals(1, requestSpecs.size()); + AdminApiDriver.RequestSpec requestSpec = requestSpecs.get(0); + + driver.onFailure(time.milliseconds(), requestSpec, new UnknownServerException()); + assertTrue(future.isDone()); + TestUtils.assertFutureThrows(future, UnknownServerException.class); + assertEquals(Collections.emptyList(), driver.poll()); + } + + private MetadataResponse responseWithBrokers(Set brokerIds) { + MetadataResponseData response = new MetadataResponseData(); + for (Integer brokerId : brokerIds) { + response.brokers().add(new MetadataResponseData.MetadataResponseBroker() + .setNodeId(brokerId) + .setHost("host" + brokerId) + .setPort(9092) + ); + } + return new MetadataResponse(response, ApiKeys.METADATA.latestVersion()); + } + + private class MockApiHandler implements AdminApiHandler { + private final AllBrokersStrategy allBrokersStrategy = new AllBrokersStrategy(logContext); + + @Override + public String apiName() { + return "mock-api"; + } + + @Override + public AbstractRequest.Builder buildRequest( + int brokerId, + Set keys + ) { + return new MetadataRequest.Builder(new MetadataRequestData()); + } + + @Override + public ApiResult handleResponse( + int brokerId, + Set keys, + AbstractResponse response + ) { + return ApiResult.completed(keys.iterator().next(), brokerId); + } + + @Override + public AdminApiLookupStrategy lookupStrategy() { + return allBrokersStrategy; + } + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyTest.java new file mode 100644 index 0000000000000..8e4b961394cbe --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyTest.java @@ -0,0 +1,124 @@ +/* + * 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.message.MetadataResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.requests.MetadataRequest; +import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.utils.LogContext; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashSet; +import java.util.OptionalInt; +import java.util.Set; + +import static org.apache.kafka.common.utils.Utils.mkSet; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class AllBrokersStrategyTest { + private final LogContext logContext = new LogContext(); + + @Test + public void testBuildRequest() { + AllBrokersStrategy strategy = new AllBrokersStrategy(logContext); + MetadataRequest.Builder builder = strategy.buildRequest(AllBrokersStrategy.LOOKUP_KEYS); + assertEquals(Collections.emptyList(), builder.topics()); + } + + @Test + public void testBuildRequestWithInvalidLookupKeys() { + AllBrokersStrategy strategy = new AllBrokersStrategy(logContext); + AllBrokersStrategy.BrokerKey key1 = new AllBrokersStrategy.BrokerKey(OptionalInt.empty()); + AllBrokersStrategy.BrokerKey key2 = new AllBrokersStrategy.BrokerKey(OptionalInt.of(1)); + assertThrows(IllegalArgumentException.class, () -> strategy.buildRequest(mkSet(key1))); + assertThrows(IllegalArgumentException.class, () -> strategy.buildRequest(mkSet(key2))); + assertThrows(IllegalArgumentException.class, () -> strategy.buildRequest(mkSet(key1, key2))); + + Set keys = new HashSet<>(AllBrokersStrategy.LOOKUP_KEYS); + keys.add(key2); + assertThrows(IllegalArgumentException.class, () -> strategy.buildRequest(keys)); + } + + @Test + public void testHandleResponse() { + AllBrokersStrategy strategy = new AllBrokersStrategy(logContext); + + MetadataResponseData response = new MetadataResponseData(); + response.brokers().add(new MetadataResponseData.MetadataResponseBroker() + .setNodeId(1) + .setHost("host1") + .setPort(9092) + ); + response.brokers().add(new MetadataResponseData.MetadataResponseBroker() + .setNodeId(2) + .setHost("host2") + .setPort(9092) + ); + + AdminApiLookupStrategy.LookupResult lookupResult = strategy.handleResponse( + AllBrokersStrategy.LOOKUP_KEYS, + new MetadataResponse(response, ApiKeys.METADATA.latestVersion()) + ); + + assertEquals(Collections.emptyMap(), lookupResult.failedKeys); + + Set expectedMappedKeys = mkSet( + new AllBrokersStrategy.BrokerKey(OptionalInt.of(1)), + new AllBrokersStrategy.BrokerKey(OptionalInt.of(2)) + ); + + assertEquals(expectedMappedKeys, lookupResult.mappedKeys.keySet()); + lookupResult.mappedKeys.forEach((brokerKey, brokerId) -> { + assertEquals(OptionalInt.of(brokerId), brokerKey.brokerId); + }); + } + + @Test + public void testHandleResponseWithNoBrokers() { + AllBrokersStrategy strategy = new AllBrokersStrategy(logContext); + + MetadataResponseData response = new MetadataResponseData(); + + AdminApiLookupStrategy.LookupResult lookupResult = strategy.handleResponse( + AllBrokersStrategy.LOOKUP_KEYS, + new MetadataResponse(response, ApiKeys.METADATA.latestVersion()) + ); + + assertEquals(Collections.emptyMap(), lookupResult.failedKeys); + assertEquals(Collections.emptyMap(), lookupResult.mappedKeys); + } + + @Test + public void testHandleResponseWithInvalidLookupKeys() { + AllBrokersStrategy strategy = new AllBrokersStrategy(logContext); + AllBrokersStrategy.BrokerKey key1 = new AllBrokersStrategy.BrokerKey(OptionalInt.empty()); + AllBrokersStrategy.BrokerKey key2 = new AllBrokersStrategy.BrokerKey(OptionalInt.of(1)); + MetadataResponse response = new MetadataResponse(new MetadataResponseData(), ApiKeys.METADATA.latestVersion()); + + assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(mkSet(key1), response)); + assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(mkSet(key2), response)); + assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(mkSet(key1, key2), response)); + + Set keys = new HashSet<>(AllBrokersStrategy.LOOKUP_KEYS); + keys.add(key2); + assertThrows(IllegalArgumentException.class, () -> strategy.handleResponse(keys, response)); + } + +} diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java index 34ebde7c7be48..d0ddb9960d75b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java @@ -39,6 +39,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.OptionalInt; import java.util.Set; import java.util.function.Function; import java.util.stream.Collectors; @@ -46,22 +47,18 @@ import static java.util.Arrays.asList; import static java.util.Collections.emptyList; import static java.util.Collections.emptyMap; -import static java.util.Collections.emptySet; import static java.util.Collections.singletonList; import static java.util.Collections.singletonMap; import static org.apache.kafka.common.utils.Utils.mkSet; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; public class DescribeProducersHandlerTest { private DescribeProducersHandler newHandler( - Set topicPartitions, DescribeProducersOptions options ) { return new DescribeProducersHandler( - topicPartitions, options, new LogContext() ); @@ -77,16 +74,13 @@ public void testBrokerIdSetInOptions() { ); DescribeProducersHandler handler = newHandler( - topicPartitions, new DescribeProducersOptions().brokerId(brokerId) ); - AdminApiHandler.Keys keys = handler.initializeKeys(); - assertEquals(emptySet(), keys.dynamicKeys); - assertNull(keys.lookupStrategy); - - keys.staticKeys.forEach((topicPartition, mappedBrokerId) -> { - assertEquals(brokerId, mappedBrokerId, "Unexpected brokerId for " + topicPartition); + topicPartitions.forEach(topicPartition -> { + ApiRequestScope scope = handler.lookupStrategy().lookupScope(topicPartition); + assertEquals(OptionalInt.of(brokerId), scope.destinationBrokerId(), + "Unexpected brokerId for " + topicPartition); }); } @@ -99,14 +93,14 @@ public void testBrokerIdNotSetInOptions() { ); DescribeProducersHandler handler = newHandler( - topicPartitions, new DescribeProducersOptions() ); - AdminApiHandler.Keys keys = handler.initializeKeys(); - assertEquals(emptyMap(), keys.staticKeys); - assertTrue(keys.lookupStrategy instanceof PartitionLeaderStrategy); - assertEquals(topicPartitions, keys.dynamicKeys); + topicPartitions.forEach(topicPartition -> { + ApiRequestScope scope = handler.lookupStrategy().lookupScope(topicPartition); + assertEquals(OptionalInt.empty(), scope.destinationBrokerId(), + "Unexpected brokerId for " + topicPartition); + }); } @Test @@ -118,7 +112,6 @@ public void testBuildRequest() { ); DescribeProducersHandler handler = newHandler( - topicPartitions, new DescribeProducersOptions() ); @@ -197,7 +190,7 @@ public void testFatalNotLeaderErrorIfStaticMapped() { public void testCompletedResult() { TopicPartition topicPartition = new TopicPartition("foo", 5); DescribeProducersOptions options = new DescribeProducersOptions().brokerId(1); - DescribeProducersHandler handler = newHandler(mkSet(topicPartition), options); + DescribeProducersHandler handler = newHandler(options); int brokerId = 3; PartitionResponse partitionResponse = sampleProducerState(topicPartition); @@ -244,7 +237,7 @@ private ApiResult handleResponseWithErro TopicPartition topicPartition, Errors error ) { - DescribeProducersHandler handler = newHandler(mkSet(topicPartition), options); + DescribeProducersHandler handler = newHandler(options); int brokerId = options.brokerId().orElse(3); DescribeProducersResponse response = buildResponseWithError(topicPartition, error); return handler.handleResponse(brokerId, mkSet(topicPartition), response); diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java index 04e7851e5f1a0..0db9cbb691781 100644 --- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java @@ -48,7 +48,7 @@ public void testBuildRequest() { String transactionalId3 = "baz"; Set transactionalIds = mkSet(transactionalId1, transactionalId2, transactionalId3); - DescribeTransactionsHandler handler = new DescribeTransactionsHandler(transactionalIds, logContext); + DescribeTransactionsHandler handler = new DescribeTransactionsHandler(logContext); assertLookup(handler, transactionalIds); assertLookup(handler, mkSet(transactionalId1)); @@ -62,7 +62,7 @@ public void testHandleSuccessfulResponse() { String transactionalId2 = "bar"; Set transactionalIds = mkSet(transactionalId1, transactionalId2); - DescribeTransactionsHandler handler = new DescribeTransactionsHandler(transactionalIds, logContext); + DescribeTransactionsHandler handler = new DescribeTransactionsHandler(logContext); DescribeTransactionsResponseData.TransactionState transactionState1 = sampleTransactionState1(transactionalId1); @@ -86,8 +86,7 @@ public void testHandleSuccessfulResponse() { @Test public void testHandleErrorResponse() { String transactionalId = "foo"; - Set transactionalIds = mkSet(transactionalId); - DescribeTransactionsHandler handler = new DescribeTransactionsHandler(transactionalIds, logContext); + DescribeTransactionsHandler handler = new DescribeTransactionsHandler(logContext); assertFatalError(handler, transactionalId, Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED); assertFatalError(handler, transactionalId, Errors.TRANSACTIONAL_ID_NOT_FOUND); assertFatalError(handler, transactionalId, Errors.UNKNOWN_SERVER_ERROR); diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java new file mode 100644 index 0000000000000..799eb40ce8c1b --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java @@ -0,0 +1,185 @@ +/* + * 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.clients.admin.internals.AdminApiHandler.ApiResult; +import org.apache.kafka.clients.admin.internals.AllBrokersStrategy.BrokerKey; +import org.apache.kafka.common.message.ListTransactionsResponseData; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.ListTransactionsRequest; +import org.apache.kafka.common.requests.ListTransactionsResponse; +import org.apache.kafka.common.utils.LogContext; +import org.junit.jupiter.api.Test; + +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.OptionalInt; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static java.util.Arrays.asList; +import static java.util.Collections.singleton; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class ListTransactionsHandlerTest { + private final LogContext logContext = new LogContext(); + + @Test + public void testBuildRequestWithoutFilters() { + int brokerId = 1; + BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId)); + ListTransactionsOptions options = new ListTransactionsOptions(); + ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); + ListTransactionsRequest request = handler.buildRequest(brokerId, singleton(brokerKey)).build(); + assertEquals(Collections.emptyList(), request.data().producerIdFilters()); + assertEquals(Collections.emptyList(), request.data().stateFilters()); + } + + @Test + public void testBuildRequestWithFilteredProducerId() { + int brokerId = 1; + BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId)); + long filteredProducerId = 23423L; + ListTransactionsOptions options = new ListTransactionsOptions() + .filterProducerIds(singleton(filteredProducerId)); + ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); + ListTransactionsRequest request = handler.buildRequest(brokerId, singleton(brokerKey)).build(); + assertEquals(Collections.singletonList(filteredProducerId), request.data().producerIdFilters()); + assertEquals(Collections.emptyList(), request.data().stateFilters()); + } + + @Test + public void testBuildRequestWithFilteredState() { + int brokerId = 1; + BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId)); + TransactionState filteredState = TransactionState.ONGOING; + ListTransactionsOptions options = new ListTransactionsOptions() + .filterStates(singleton(filteredState)); + ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); + ListTransactionsRequest request = handler.buildRequest(brokerId, singleton(brokerKey)).build(); + assertEquals(Collections.singletonList(filteredState.toString()), request.data().stateFilters()); + assertEquals(Collections.emptyList(), request.data().producerIdFilters()); + } + + @Test + public void testHandleSuccessfulResponse() { + int brokerId = 1; + BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId)); + ListTransactionsOptions options = new ListTransactionsOptions(); + ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); + ListTransactionsResponse response = sampleListTransactionsResponse1(); + ApiResult> result = handler.handleResponse( + brokerId, singleton(brokerKey), response); + assertEquals(singleton(brokerKey), result.completedKeys.keySet()); + assertExpectedTransactions(response.data().transactionStates(), result.completedKeys.get(brokerKey)); + } + + @Test + public void testCoordinatorLoadingErrorIsRetriable() { + int brokerId = 1; + ApiResult> result = + handleResponseWithError(brokerId, Errors.COORDINATOR_LOAD_IN_PROGRESS); + assertEquals(Collections.emptyMap(), result.completedKeys); + assertEquals(Collections.emptyMap(), result.failedKeys); + assertEquals(Collections.emptyList(), result.unmappedKeys); + } + + @Test + public void testHandleResponseWithFatalErrors() { + assertFatalError(Errors.COORDINATOR_NOT_AVAILABLE); + assertFatalError(Errors.UNKNOWN_SERVER_ERROR); + } + + private void assertFatalError( + Errors error + ) { + int brokerId = 1; + BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId)); + ApiResult> result = handleResponseWithError(brokerId, error); + assertEquals(Collections.emptyMap(), result.completedKeys); + assertEquals(Collections.emptyList(), result.unmappedKeys); + assertEquals(Collections.singleton(brokerKey), result.failedKeys.keySet()); + Throwable throwable = result.failedKeys.get(brokerKey); + assertEquals(error, Errors.forException(throwable)); + } + + private ApiResult> handleResponseWithError( + int brokerId, + Errors error + ) { + BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId)); + ListTransactionsOptions options = new ListTransactionsOptions(); + ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext); + + ListTransactionsResponse response = new ListTransactionsResponse( + new ListTransactionsResponseData().setErrorCode(error.code()) + ); + return handler.handleResponse(brokerId, singleton(brokerKey), response); + } + + private ListTransactionsResponse sampleListTransactionsResponse1() { + return new ListTransactionsResponse( + new ListTransactionsResponseData() + .setErrorCode(Errors.NONE.code()) + .setTransactionStates(asList( + new ListTransactionsResponseData.TransactionState() + .setTransactionalId("foo") + .setProducerId(12345L) + .setTransactionState("Ongoing"), + new ListTransactionsResponseData.TransactionState() + .setTransactionalId("bar") + .setProducerId(98765L) + .setTransactionState("PrepareAbort") + )) + ); + } + + private void assertExpectedTransactions( + List expected, + Collection actual + ) { + assertEquals(expected.size(), actual.size()); + + Map expectedMap = expected.stream().collect(Collectors.toMap( + ListTransactionsResponseData.TransactionState::transactionalId, + Function.identity() + )); + + for (TransactionListing actualListing : actual) { + ListTransactionsResponseData.TransactionState expectedState = + expectedMap.get(actualListing.transactionalId()); + assertNotNull(expectedState); + assertExpectedTransactionState(expectedState, actualListing); + } + } + + private void assertExpectedTransactionState( + ListTransactionsResponseData.TransactionState expected, + TransactionListing actual + ) { + assertEquals(expected.transactionalId(), actual.transactionalId()); + assertEquals(expected.producerId(), actual.producerId()); + assertEquals(expected.transactionState(), actual.state().toString()); + } + +}