kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cmcc...@apache.org
Subject [kafka] branch trunk updated: KAFKA-7808: AdminClient#describeTopics should not throw InvalidTopic if topic name is not found (#6124)
Date Fri, 11 Jan 2019 22:56:20 GMT
This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 7df3e8c  KAFKA-7808: AdminClient#describeTopics should not throw InvalidTopic if
topic name is not found (#6124)
7df3e8c is described below

commit 7df3e8cd38083ae78ec1e26a259cfb3bb1fbe2d2
Author: Lee Dongjin <dongjin@apache.org>
AuthorDate: Sat Jan 12 07:56:06 2019 +0900

    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>
---
 .../kafka/clients/admin/KafkaAdminClient.java      |  3 +-
 .../kafka/clients/admin/MockAdminClient.java       |  3 +-
 .../apache/kafka/trogdor/common/WorkerUtils.java   | 32 ++++------------------
 3 files changed, 9 insertions(+), 29 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 39817df..ce51eee 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/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);
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
index 9fe1ba4..ed7efd5 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++ b/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);
             }
         }
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java b/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java
index 3d4871a..ef6e275 100644
--- a/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java
+++ b/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;
@@ -135,34 +136,11 @@ 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'
      */


Mime
View raw message