KAFKA-12709; Add Admin API for ListTransactions

clients/src/main/java/org/apache/kafka/clients/admin/AbortTransactionOptions.java

@@ -20,4 +20,12 @@
 
 @InterfaceStability.Evolving
 public class AbortTransactionOptions extends AbstractOptions<AbortTransactionOptions> {
+
+    @Override
+    public String toString() {
+        return "AbortTransactionOptions(" +
+            "timeoutMs=" + timeoutMs +
+            ')';
+    }
+
 }

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
      */

clients/src/main/java/org/apache/kafka/clients/admin/DescribeProducersResult.java

@@ -71,6 +71,13 @@ public PartitionProducerState(List<ProducerState> activeProducers) {
         public List<ProducerState> activeProducers() {
             return activeProducers;
         }
+
+        @Override
+        public String toString() {
+            return "PartitionProducerState(" +
+                "activeProducers=" + activeProducers +
+                ')';
+        }
     }
 
 }

clients/src/main/java/org/apache/kafka/clients/admin/DescribeTransactionsOptions.java

@@ -28,4 +28,11 @@
 @InterfaceStability.Evolving
 public class DescribeTransactionsOptions extends AbstractOptions<DescribeTransactionsOptions> {
 
+    @Override
+    public String toString() {
+        return "DescribeTransactionsOptions(" +
+            "timeoutMs=" + timeoutMs +
+            ')';
+    }
+
 }

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,48 +4733,57 @@ void handleFailure(Throwable throwable) {
 
     @Override
     public DescribeProducersResult describeProducers(Collection<TopicPartition> topicPartitions, DescribeProducersOptions options) {
-        DescribeProducersHandler handler = new DescribeProducersHandler(
-            new HashSet<>(topicPartitions),
-            options,
-            logContext
-        );
-        return new DescribeProducersResult(invokeDriver(handler, options.timeoutMs));
+        AdminApiFuture.SimpleAdminApiFuture<TopicPartition, DescribeProducersResult.PartitionProducerState> 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<String> transactionalIds, DescribeTransactionsOptions options) {
-        DescribeTransactionsHandler handler = new DescribeTransactionsHandler(
-            transactionalIds,
-            logContext
-        );
-        return new DescribeTransactionsResult(invokeDriver(handler, options.timeoutMs));
+        AdminApiFuture.SimpleAdminApiFuture<CoordinatorKey, TransactionDescription> 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<TopicPartition, Void> 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<Collection<TransactionListing>> future =
+            ListTransactionsHandler.newFuture();
+        ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
+        invokeDriver(handler, future, options.timeoutMs);
+        return new ListTransactionsResult(future.all());
     }
 
-    private <K, V> Map<K, KafkaFutureImpl<V>> invokeDriver(
+    private <K, V> void invokeDriver(
         AdminApiHandler<K, V> handler,
+        AdminApiFuture<K, V> future,
         Integer timeoutMs
     ) {
         long currentTimeMs = time.milliseconds();
         long deadlineMs = calcDeadlineMs(currentTimeMs, timeoutMs);
 
         AdminApiDriver<K, V> driver = new AdminApiDriver<>(
             handler,
+            future,
             deadlineMs,
             retryBackoffMs,
             logContext
         );
 
         maybeSendRequests(driver, currentTimeMs);
-        return driver.futures();
     }
 
     private <K, V> void maybeSendRequests(AdminApiDriver<K, V> driver, long currentTimeMs) {

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<ListTransactionsOptions> {
+    private Set<TransactionState> filteredStates = Collections.emptySet();
+    private Set<Long> 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<TransactionState> 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<Long> 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<TransactionState> 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<Long> filteredProducerIds() {
+        return filteredProducerIds;
+    }
+
+    @Override
+    public String toString() {
+        return "ListTransactionsOptions(" +
+            "filteredStates=" + filteredStates +
+            ", filteredProducerIds=" + filteredProducerIds +
+            ", timeoutMs=" + timeoutMs +
+            ')';
+    }
+}

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.
+ * <p>
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class ListTransactionsResult {
+    private final KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> future;
+
+    ListTransactionsResult(KafkaFutureImpl<Map<Integer, KafkaFutureImpl<Collection<TransactionListing>>>> 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<Collection<TransactionListing>> all() {
+        return allByBrokerId().thenApply(map -> {
+            List<TransactionListing> allListings = new ArrayList<>();
+            for (Collection<TransactionListing> 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<Map<Integer, KafkaFuture<Collection<TransactionListing>>>> byBrokerId() {
+        KafkaFutureImpl<Map<Integer, KafkaFuture<Collection<TransactionListing>>>> result = new KafkaFutureImpl<>();
+        future.whenComplete((brokerFutures, exception) -> {
+            if (brokerFutures != null) {
+                Map<Integer, KafkaFuture<Collection<TransactionListing>>> 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<Map<Integer, Collection<TransactionListing>>> allByBrokerId() {
+        KafkaFutureImpl<Map<Integer, Collection<TransactionListing>>> allFuture = new KafkaFutureImpl<>();
+        Map<Integer, Collection<TransactionListing>> allListingsMap = new HashMap<>();
+
+        future.whenComplete((map, topLevelException) -> {
+            if (topLevelException != null) {
+                allFuture.completeExceptionally(topLevelException);
+                return;
+            }
+
+            Set<Integer> 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;
+    }
+
+}