[For Review] Pt. 2 - Introduce TopologyMetadata to wrap InternalTopologyBuilders of named topologies

build.gradle

@@ -30,7 +30,7 @@ buildscript {
 }
 
 plugins {
-  id 'com.diffplug.spotless' version '5.12.4'
+  id 'com.diffplug.spotless' version '5.12.5'
   id 'com.github.ben-manes.versions' version '0.38.0'
   id 'idea'
   id 'java-library'
@@ -449,6 +449,14 @@ subprojects {
     }
   }
 
+  // remove test output from all test types
+  tasks.withType(Test).all { t ->
+    cleanTest {
+      delete t.reports.junitXml.destination
+      delete t.reports.html.destination
+    }
+  }
+
   jar {
     from "$rootDir/LICENSE"
     from "$rootDir/NOTICE"

checkstyle/suppressions.xml

@@ -152,7 +152,7 @@
 
     <!-- Streams -->
     <suppress checks="ClassFanOutComplexity"
-              files="(KafkaStreams|KStreamImpl|KTableImpl|StreamsPartitionAssignor).java"/>
+              files="(KafkaStreams|KStreamImpl|KTableImpl|InternalTopologyBuilder|StreamsPartitionAssignor).java"/>
 
     <suppress checks="MethodLength"
               files="KTableImpl.java"/>

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

@@ -338,7 +338,7 @@ default DescribeClusterResult describeCluster() {
      * This operation is supported by brokers with version 0.11.0.0 or higher.
      *
      * @param filter The filter to use.
-     * @return The DeleteAclsResult.
+     * @return The DescribeAclsResult.
      */
     default DescribeAclsResult describeAcls(AclBindingFilter filter) {
         return describeAcls(filter, new DescribeAclsOptions());
@@ -354,7 +354,7 @@ default DescribeAclsResult describeAcls(AclBindingFilter filter) {
      *
      * @param filter  The filter to use.
      * @param options The options to use when listing the ACLs.
-     * @return The DeleteAclsResult.
+     * @return The DescribeAclsResult.
      */
     DescribeAclsResult describeAcls(AclBindingFilter filter, DescribeAclsOptions options);
 
@@ -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 +
+            ')';
+    }
+}