Minor: Don’t send the DeleteTopicsRequest for the invalid topic names

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

@@ -1150,9 +1150,10 @@ void handleFailure(Throwable throwable) {
     }
 
     @Override
-    public DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
+    public DeleteTopicsResult deleteTopics(Collection<String> topicNames,
                                            DeleteTopicsOptions options) {
         final Map<String, KafkaFutureImpl<Void>> topicFutures = new HashMap<>(topicNames.size());
+        final List<String> validTopicNames = new ArrayList<>(topicNames.size());
         for (String topicName : topicNames) {
             if (topicNameIsUnrepresentable(topicName)) {
                 KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();
@@ -1161,6 +1162,7 @@ public DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
                 topicFutures.put(topicName, future);
             } else if (!topicFutures.containsKey(topicName)) {
                 topicFutures.put(topicName, new KafkaFutureImpl<Void>());
+                validTopicNames.add(topicName);
             }
         }
         final long now = time.milliseconds();
@@ -1169,7 +1171,7 @@ public DeleteTopicsResult deleteTopics(final Collection<String> topicNames,
 
             @Override
             AbstractRequest.Builder createRequest(int timeoutMs) {
-                return new DeleteTopicsRequest.Builder(new HashSet<>(topicNames), timeoutMs);
+                return new DeleteTopicsRequest.Builder(new HashSet<>(validTopicNames), timeoutMs);
             }
 
             @Override
@@ -1204,7 +1206,7 @@ void handleFailure(Throwable throwable) {
                 completeAllExceptionally(topicFutures.values(), throwable);
             }
         };
-        if (!topicNames.isEmpty()) {
+        if (!validTopicNames.isEmpty()) {
             runnable.call(call, now);
         }
         return new DeleteTopicsResult(new HashMap<String, KafkaFuture<Void>>(topicFutures));

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.clients;
 
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.errors.AuthenticationException;
@@ -88,7 +89,7 @@ public FutureResponse(Node node,
     private final Queue<MetadataUpdate> metadataUpdates = new ArrayDeque<>();
     private volatile NodeApiVersions nodeApiVersions = NodeApiVersions.create();
     private volatile int numBlockingWakeups = 0;
-
+    private final AtomicInteger totalRequestCount = new AtomicInteger(0);
     public MockClient(Time time) {
         this(time, null);
     }
@@ -394,6 +395,7 @@ public void reset() {
         futureResponses.clear();
         metadataUpdates.clear();
         authenticationErrors.clear();
+        totalRequestCount.set(0);
     }
 
     public boolean hasPendingMetadataUpdates() {
@@ -461,6 +463,7 @@ public ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder<?>
     @Override
     public ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder<?> requestBuilder, long createdTimeMs,
                                           boolean expectResponse, RequestCompletionHandler callback) {
+        totalRequestCount.incrementAndGet();
         return new ClientRequest(nodeId, requestBuilder, 0, "mockClientId", createdTimeMs,
                 expectResponse, callback);
     }
@@ -503,4 +506,9 @@ private static class MetadataUpdate {
             this.expectMatchRefreshTopics = expectMatchRefreshTopics;
         }
     }
+
+    // visible for testing
+    public int totalRequestCount() {
+        return totalRequestCount.get();
+    }
 }

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

@@ -230,22 +230,26 @@ public void testInvalidTopicNames() throws Exception {
             for (String sillyTopicName : sillyTopicNames) {
                 assertFutureError(deleteFutures.get(sillyTopicName), InvalidTopicException.class);
             }
+            assertEquals(0, env.kafkaClient().totalRequestCount());
 
             Map<String, KafkaFuture<TopicDescription>> describeFutures =
                     env.adminClient().describeTopics(sillyTopicNames).values();
             for (String sillyTopicName : sillyTopicNames) {
                 assertFutureError(describeFutures.get(sillyTopicName), InvalidTopicException.class);
             }
+            assertEquals(0, env.kafkaClient().totalRequestCount());
 
             List<NewTopic> newTopics = new ArrayList<>();
             for (String sillyTopicName : sillyTopicNames) {
                 newTopics.add(new NewTopic(sillyTopicName, 1, (short) 1));
             }
+
             Map<String, KafkaFuture<Void>> createFutures =
                 env.adminClient().createTopics(newTopics).values();
             for (String sillyTopicName : sillyTopicNames) {
                 assertFutureError(createFutures .get(sillyTopicName), InvalidTopicException.class);
             }
+            assertEquals(0, env.kafkaClient().totalRequestCount());
         }
     }