Merge branch 'axbaretto'

* axbaretto:
  KAFKA-12156: Document single threaded response handling in Admin client (#9842)

Author: a0x8o

build.gradle

@@ -241,7 +241,7 @@ subprojects {
     }
   }
 
-  def shouldUseJUnit5 = ["tools", "raft"].contains(it.project.name)
+  def shouldUseJUnit5 = ["tools", "raft", "log4j-appender"].contains(it.project.name)
   def testLoggingEvents = ["passed", "skipped", "failed"]
   def testShowStandardStreams = false
   def testExceptionFormat = 'full'
@@ -1157,6 +1157,7 @@ project(':clients') {
     include "**/org/apache/kafka/common/header/*"
     include "**/org/apache/kafka/common/metrics/*"
     include "**/org/apache/kafka/common/metrics/stats/*"
+    include "**/org/apache/kafka/common/quota/*"
     include "**/org/apache/kafka/common/resource/*"
     include "**/org/apache/kafka/common/serialization/*"
     include "**/org/apache/kafka/common/config/*"
@@ -1782,8 +1783,8 @@ project(':log4j-appender') {
     compile libs.slf4jlog4j
 
     testCompile project(':clients').sourceSets.test.output
-    testCompile libs.junitJupiterApi
-    testCompile libs.junitVintageEngine
+    testCompile libs.junitJupiter
+    testCompile libs.hamcrest
     testCompile libs.easymock
   }
 

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -1171,7 +1171,7 @@ private InProgressData(int requestVersion, boolean isPartialUpdate) {
                 this.requestVersion = requestVersion;
                 this.isPartialUpdate = isPartialUpdate;
             }
-        };
+        }
 
     }
 

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

@@ -29,6 +29,7 @@
 
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.ElectionType;
+import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
@@ -46,6 +47,10 @@
  * The administrative client for Kafka, which supports managing and inspecting topics, brokers, configurations and ACLs.
  * <p>
  * Instances returned from the {@code create} methods of this interface are guaranteed to be thread safe.
+ * However, the {@link KafkaFuture KafkaFutures} returned from request methods are executed
+ * by a single thread so it is important that any code which executes on that thread when they complete
+ * (using {@link KafkaFuture#thenApply(KafkaFuture.Function)}, for example) doesn't block
+ * for too long. If necessary, processing of results should be passed to another thread.
  * <p>
  * The operations exposed by Admin follow a consistent pattern:
  * <ul>
@@ -57,11 +62,11 @@
  *     preferred over multiple calls to the same method.
  *     <li>The operation methods execute asynchronously.
  *     <li>Each {@code xxx} operation method returns an {@code XxxResult} class with methods which expose
- *     {@link org.apache.kafka.common.KafkaFuture} for accessing the result(s) of the operation.
+ *     {@link KafkaFuture} for accessing the result(s) of the operation.
  *     <li>Typically an {@code all()} method is provided for getting the overall success/failure of the batch and a
  *     {@code values()} method provided access to each item in a request batch.
  *     Other methods may also be provided.
- *     <li>For synchronous behaviour use {@link org.apache.kafka.common.KafkaFuture#get()}
+ *     <li>For synchronous behaviour use {@link KafkaFuture#get()}
  * </ul>
  * <p>
  * Here is a simple example of using an Admin client instance to create a new topic:

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -108,17 +108,18 @@ public class ConsumerConfig extends AbstractConfig {
      */
     public static final String PARTITION_ASSIGNMENT_STRATEGY_CONFIG = "partition.assignment.strategy";
     private static final String PARTITION_ASSIGNMENT_STRATEGY_DOC = "A list of class names or class types, " +
-            "ordered by preference, of supported partition assignment " +
-            "strategies that the client will use to distribute partition " +
-            "ownership amongst consumer instances when group management is " +
-            "used.<p>In addition to the default class specified below, " +
-            "you can use the " +
-            "<code>org.apache.kafka.clients.consumer.RoundRobinAssignor</code>" +
-            "class for round robin assignments of partitions to consumers. " +
-            "</p><p>Implementing the " +
-            "<code>org.apache.kafka.clients.consumer.ConsumerPartitionAssignor" +
-            "</code> interface allows you to plug in a custom assignment" +
-            "strategy.";
+        "ordered by preference, of supported partition assignment strategies that the client will use to distribute " +
+        "partition ownership amongst consumer instances when group management is used. Available options are:" +
+        "<ul>" +
+        "<li><code>org.apache.kafka.clients.consumer.RangeAssignor</code>: The default assignor, which works on a per-topic basis.</li>" +
+        "<li><code>org.apache.kafka.clients.consumer.RoundRobinAssignor</code>: Assigns partitions to consumers in a round-robin fashion.</li>" +
+        "<li><code>org.apache.kafka.clients.consumer.StickyAssignor</code>: Guarantees an assignment that is " +
+        "maximally balanced while preserving as many existing partition assignments as possible.</li>" +
+        "<li><code>org.apache.kafka.clients.consumer.CooperativeStickyAssignor</code>: Follows the same StickyAssignor " +
+        "logic, but allows for cooperative rebalancing.</li>" +
+        "</ul>" +
+        "<p>Implementing the <code>org.apache.kafka.clients.consumer.ConsumerPartitionAssignor</code> " +
+        "interface allows you to plug in a custom assignment strategy.";
 
     /**
      * <code>auto.offset.reset</code>

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionalRequestResult.java

@@ -71,7 +71,7 @@ public void await(long timeout, TimeUnit unit) {
                 throw error();
             }
             if (!success) {
-                throw new TimeoutException("Timeout expired after " + timeout + unit.name().toLowerCase(Locale.ROOT) + " while awaiting " + operation);
+                throw new TimeoutException("Timeout expired after " + timeout + " " + unit.name().toLowerCase(Locale.ROOT) + " while awaiting " + operation);
             }
         } catch (InterruptedException e) {
             throw new InterruptException("Received interrupt while awaiting " + operation, e);

clients/src/main/java/org/apache/kafka/common/Cluster.java

@@ -232,9 +232,9 @@ public List<Node> nodes() {
     }
 
     /**
-     * Get the node by the node id (or null if no such node exists)
+     * Get the node by the node id (or null if the node is not online or does not exist)
      * @param id The id of the node
-     * @return The node, or null if no such node exists
+     * @return The node, or null if the node is not online or does not exist
      */
     public Node nodeById(int id) {
         return this.nodesById.get(id);

clients/src/main/resources/common/message/OffsetDeleteRequest.json

@@ -19,11 +19,11 @@
   "name": "OffsetDeleteRequest",
   "validVersions": "0",
   "fields": [
-    { "name": "GroupId", "type": "string", "versions": "0+",
+    { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
       "about": "The unique group identifier." },
     { "name": "Topics", "type": "[]OffsetDeleteRequestTopic", "versions": "0+",
       "about": "The topics to delete offsets for", "fields": [
-        { "name": "Name",  "type": "string",  "versions": "0+", "mapKey": true,
+        { "name": "Name",  "type": "string",  "versions": "0+", "mapKey": true, "entityType": "topicName",
           "about": "The topic name." },
         { "name": "Partitions", "type": "[]OffsetDeleteRequestPartition", "versions": "0+",
           "about": "Each partition to delete offsets for.", "fields": [

clients/src/main/resources/common/message/OffsetDeleteResponse.json

@@ -25,7 +25,7 @@
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
     { "name": "Topics", "type": "[]OffsetDeleteResponseTopic", "versions": "0+",
       "about": "The responses for each topic.", "fields": [
-        { "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
+        { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
           "about": "The topic name." },
         { "name": "Partitions", "type": "[]OffsetDeleteResponsePartition", "versions": "0+",
           "about": "The responses for each partition in the topic.", "fields": [

clients/src/main/resources/common/message/TxnOffsetCommitRequest.json

@@ -25,7 +25,7 @@
   "validVersions": "0-3",
   "flexibleVersions": "3+",
   "fields": [
-    { "name": "TransactionalId", "type": "string", "versions": "0+",
+    { "name": "TransactionalId", "type": "string", "versions": "0+", "entityType": "transactionalId",
       "about": "The ID of the transaction." },
     { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
       "about": "The ID of the group." },

clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java

@@ -20,11 +20,13 @@
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.config.ConfigException;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import org.junit.Test;
 
@@ -62,19 +64,20 @@ public void testParseAndValidateAddressesWithReverseLookup() {
         validatedAddresses.forEach(address -> assertEquals(10000, address.getPort()));
     }
 
-    @Test(expected = IllegalArgumentException.class)
+    @Test
     public void testInvalidConfig() {
-        ClientUtils.parseAndValidateAddresses(Arrays.asList("localhost:10000"), "random.value");
+        assertThrows(IllegalArgumentException.class,
+            () -> ClientUtils.parseAndValidateAddresses(Collections.singletonList("localhost:10000"), "random.value"));
     }
 
-    @Test(expected = ConfigException.class)
+    @Test
     public void testNoPort() {
-        checkWithoutLookup("127.0.0.1");
+        assertThrows(ConfigException.class, () -> checkWithoutLookup("127.0.0.1"));
     }
 
-    @Test(expected = ConfigException.class)
+    @Test
     public void testOnlyBadHostname() {
-        checkWithoutLookup("some.invalid.hostname.foo.bar.local:9999");
+        assertThrows(ConfigException.class, () -> checkWithoutLookup("some.invalid.hostname.foo.bar.local:9999"));
     }
 
     @Test
@@ -95,9 +98,10 @@ public void testFilterPreferredAddresses() throws UnknownHostException {
         assertEquals(1, result.size());
     }
 
-    @Test(expected = UnknownHostException.class)
-    public void testResolveUnknownHostException() throws UnknownHostException {
-        ClientUtils.resolve("some.invalid.hostname.foo.bar.local", ClientDnsLookup.USE_ALL_DNS_IPS);
+    @Test
+    public void testResolveUnknownHostException() {
+        assertThrows(UnknownHostException.class,
+            () -> ClientUtils.resolve("some.invalid.hostname.foo.bar.local", ClientDnsLookup.USE_ALL_DNS_IPS));
     }
 
     @Test

clients/src/test/java/org/apache/kafka/clients/InFlightRequestsTest.java

@@ -30,6 +30,7 @@
 import java.util.List;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
 
 public class InFlightRequestsTest {
 
@@ -99,14 +100,14 @@ public void testCompleteNext() {
         assertEquals(0, inFlightRequests.count());
     }
 
-    @Test(expected = IllegalStateException.class)
+    @Test
     public void testCompleteNextThrowsIfNoInflights() {
-        inFlightRequests.completeNext(dest);
+        assertThrows(IllegalStateException.class, () -> inFlightRequests.completeNext(dest));
     }
 
-    @Test(expected = IllegalStateException.class)
+    @Test
     public void testCompleteLastSentThrowsIfNoInFlights() {
-        inFlightRequests.completeLastSent(dest);
+        assertThrows(IllegalStateException.class, () -> inFlightRequests.completeLastSent(dest));
     }
 
     private int addRequest(String destination) {

clients/src/test/java/org/apache/kafka/clients/MetadataTest.java

@@ -76,10 +76,10 @@ private static MetadataResponse emptyMetadataResponse() {
                 Collections.emptyList());
     }
 
-    @Test(expected = IllegalStateException.class)
+    @Test
     public void testMetadataUpdateAfterClose() {
         metadata.close();
-        metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 1000);
+        assertThrows(IllegalStateException.class, () -> metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 1000));
     }
 
     private static void checkTimeToNextUpdate(long refreshBackoffMs, long metadataExpireMs) {

clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java

@@ -63,6 +63,7 @@
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public class NetworkClientTest {
@@ -119,13 +120,12 @@ public void setup() {
         selector.reset();
     }
 
-    @Test(expected = IllegalStateException.class)
+    @Test
     public void testSendToUnreadyNode() {
         MetadataRequest.Builder builder = new MetadataRequest.Builder(Collections.singletonList("test"), true);
         long now = time.milliseconds();
         ClientRequest request = client.newClientRequest("5", builder, now, false);
-        client.send(request, now);
-        client.poll(1, time.milliseconds());
+        assertThrows(IllegalStateException.class, () -> client.send(request, now));
     }
 
     @Test

clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java

@@ -28,6 +28,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 public class NodeApiVersionsTest {
@@ -103,28 +104,32 @@ public void testLatestUsableVersion() {
         assertEquals(3, apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 3, (short) 4));
     }
 
-    @Test(expected = UnsupportedVersionException.class)
+    @Test
     public void testLatestUsableVersionOutOfRangeLow() {
         NodeApiVersions apiVersions = NodeApiVersions.create(ApiKeys.PRODUCE.id, (short) 1, (short) 2);
-        apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 3, (short) 4);
+        assertThrows(UnsupportedVersionException.class,
+            () -> apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 3, (short) 4));
     }
 
-    @Test(expected = UnsupportedVersionException.class)
+    @Test
     public void testLatestUsableVersionOutOfRangeHigh() {
         NodeApiVersions apiVersions = NodeApiVersions.create(ApiKeys.PRODUCE.id, (short) 2, (short) 3);
-        apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 0, (short) 1);
+        assertThrows(UnsupportedVersionException.class,
+            () -> apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 0, (short) 1));
     }
 
-    @Test(expected = UnsupportedVersionException.class)
+    @Test
     public void testUsableVersionCalculationNoKnownVersions() {
         NodeApiVersions versions = new NodeApiVersions(new ApiVersionsResponseKeyCollection());
-        versions.latestUsableVersion(ApiKeys.FETCH);
+        assertThrows(UnsupportedVersionException.class,
+            () -> versions.latestUsableVersion(ApiKeys.FETCH));
     }
 
-    @Test(expected = UnsupportedVersionException.class)
+    @Test
     public void testLatestUsableVersionOutOfRange() {
         NodeApiVersions apiVersions = NodeApiVersions.create(ApiKeys.PRODUCE.id, (short) 300, (short) 300);
-        apiVersions.latestUsableVersion(ApiKeys.PRODUCE);
+        assertThrows(UnsupportedVersionException.class,
+            () -> apiVersions.latestUsableVersion(ApiKeys.PRODUCE));
     }
 
     @Test

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -1067,7 +1067,7 @@ Errors.NONE, new TopicPartition(topic, 0), Optional.of(leader.id()), Optional.of
             DescribeTopicsResult result = env.adminClient().describeTopics(Collections.singleton(topic));
             Map<String, TopicDescription> topicDescriptions = result.all().get();
             assertEquals(leader, topicDescriptions.get(topic).partitions().get(0).leader());
-            assertEquals(null, topicDescriptions.get(topic).authorizedOperations());
+            assertNull(topicDescriptions.get(topic).authorizedOperations());
         }
     }
 
@@ -1247,9 +1247,9 @@ public void testDeleteAcls() throws Exception {
             DeleteAclsResult results = env.adminClient().deleteAcls(asList(FILTER1, FILTER2));
             Map<AclBindingFilter, KafkaFuture<FilterResults>> filterResults = results.values();
             FilterResults filter1Results = filterResults.get(FILTER1).get();
-            assertEquals(null, filter1Results.values().get(0).exception());
+            assertNull(filter1Results.values().get(0).exception());
             assertEquals(ACL1, filter1Results.values().get(0).binding());
-            assertEquals(null, filter1Results.values().get(1).exception());
+            assertNull(filter1Results.values().get(1).exception());
             assertEquals(ACL2, filter1Results.values().get(1).binding());
             TestUtils.assertFutureError(filterResults.get(FILTER2), SecurityDisabledException.class);
             TestUtils.assertFutureError(results.all(), SecurityDisabledException.class);
@@ -2068,7 +2068,7 @@ public void testDescribeCluster() throws Exception {
             final DescribeClusterResult result = env.adminClient().describeCluster();
             assertEquals(env.cluster().clusterResource().clusterId(), result.clusterId().get());
             assertEquals(2, result.controller().get().id());
-            assertEquals(null, result.authorizedOperations().get());
+            assertNull(result.authorizedOperations().get());
 
             // Test DescribeCluster with the authorized operations included.
             final DescribeClusterResult result2 = env.adminClient().describeCluster();