KAFKA-7808: AdminClient#describeTopics should not throw InvalidTopic if topic name is not found (#6124)

* Update KafkaAdminClient#describeTopics to throw UnknownTopicOrPartitionException.

* Remove unused method: WorkerUtils#getMatchingTopicPartitions.

* Add some JavaDoc.

Reviewed-by: Colin P. McCabe <cmccabe@apache.org>, Ryanne Dolan <ryannedolan@gmail.com>

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

@@ -56,6 +56,7 @@ import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.UnknownServerException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.internals.KafkaFutureImpl;
 import org.apache.kafka.common.metrics.JmxReporter;
@@ -1461,7 +1462,7 @@ public class KafkaAdminClient extends AdminClient {
                         continue;
                     }
                     if (!cluster.topics().contains(topicName)) {
-                        future.completeExceptionally(new InvalidTopicException("Topic " + topicName + " not found."));
+                        future.completeExceptionally(new UnknownTopicOrPartitionException("Topic " + topicName + " not found."));
                         continue;
                     }
                     boolean isInternal = cluster.internalTopics().contains(topicName);

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

@@ -235,8 +235,7 @@ public class MockAdminClient extends AdminClient {
             }
             if (!topicDescriptions.containsKey(requestedTopic)) {
                 KafkaFutureImpl<TopicDescription> future = new KafkaFutureImpl<>();
-                future.completeExceptionally(new UnknownTopicOrPartitionException(
-                    String.format("Topic %s unknown.", requestedTopic)));
+                future.completeExceptionally(new UnknownTopicOrPartitionException("Topic " + requestedTopic + " not found."));
                 topicDescriptions.put(requestedTopic, future);
             }
         }

tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java

@@ -32,6 +32,7 @@ import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.errors.NotEnoughReplicasException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.internals.KafkaFutureImpl;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
@@ -134,35 +135,12 @@ public final class WorkerUtils {
         }
     }
 
-    /**
-     * Returns a list of all existing topic partitions  that match the following criteria: topic
-     * name matches give regular expression 'topicRegex', topic is not internal, partitions are
-     * in range [startPartition, endPartition]
-     *
-     * @param log                The logger to use.
-     * @param bootstrapServers   The bootstrap server list.
-     * @param topicRegex         Topic name regular expression
-     * @param startPartition     Starting partition of partition range
-     * @param endPartition       Ending partition of partition range
-     * @return List of topic partitions
-     * @throws Throwable If getting list of topics or their descriptions fails.
-     */
-    public static Collection<TopicPartition> getMatchingTopicPartitions(
-        Logger log, String bootstrapServers,
-        Map<String, String> commonClientConf, Map<String, String> adminClientConf,
-        String topicRegex, int startPartition, int endPartition) throws Throwable {
-        try (AdminClient adminClient
-                 = createAdminClient(bootstrapServers, commonClientConf, adminClientConf)) {
-            return getMatchingTopicPartitions(adminClient, topicRegex, startPartition, endPartition);
-        } catch (Exception e) {
-            log.warn("Failed to get topic partitions matching {}", topicRegex, e);
-            throw e;
-        }
-    }
-
     /**
      * The actual create topics functionality is separated into this method and called from the
      * above method to be able to unit test with mock adminClient.
+     * @throws TopicExistsException if the specified topic already exists.
+     * @throws UnknownTopicOrPartitionException if topic creation was issued but failed to verify if it was created.
+     * @throws Throwable if creation of one or more topics fails (except for the cases above).
      */
     static void createTopics(
         Logger log, AdminClient adminClient,
@@ -258,6 +236,8 @@ public final class WorkerUtils {
      * @param topicsToVerify     List of topics to verify
      * @param topicsInfo         Map of topic name to topic description, which includes topics in
      *                           'topicsToVerify' list.
+     * @throws UnknownTopicOrPartitionException If at least one topic contained in 'topicsInfo'
+     * does not exist
      * @throws RuntimeException  If one or more topics have different number of partitions than
      * described in 'topicsInfo'
      */