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: MINOR: WorkerUtils#topicDescriptions must unwrap exceptions properly (#6937)
Date Wed, 03 Jul 2019 23:08:55 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 822abe4  MINOR: WorkerUtils#topicDescriptions must unwrap exceptions properly (#6937)
822abe4 is described below

commit 822abe47db07fcf92e8aa24e920a54239a90348d
Author: Colin Patrick McCabe <cmccabe@apache.org>
AuthorDate: Wed Jul 3 16:08:39 2019 -0700

    MINOR: WorkerUtils#topicDescriptions must unwrap exceptions properly (#6937)
    
    Reviewers: Ismael Juma <ismael@juma.me.uk>, Stanislav Kozlovski <stanislav_kozlovski@outlook.com>
---
 .../kafka/clients/admin/MockAdminClient.java       | 32 +++++++++++++++-----
 .../apache/kafka/trogdor/common/WorkerUtils.java   | 12 +++++---
 .../kafka/trogdor/common/WorkerUtilsTest.java      | 35 +++++++++++++++-------
 3 files changed, 58 insertions(+), 21 deletions(-)

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 19f9eae..7ca9ce4 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
@@ -205,7 +205,11 @@ public class MockAdminClient extends AdminClient {
 
         for (Map.Entry<String, TopicMetadata> topicDescription : allTopics.entrySet())
{
             String topicName = topicDescription.getKey();
-            topicListings.put(topicName, new TopicListing(topicName, topicDescription.getValue().isInternalTopic));
+            if (topicDescription.getValue().fetchesRemainingUntilVisible > 0) {
+                topicDescription.getValue().fetchesRemainingUntilVisible--;
+            } else {
+                topicListings.put(topicName, new TopicListing(topicName, topicDescription.getValue().isInternalTopic));
+            }
         }
 
         KafkaFutureImpl<Map<String, TopicListing>> future = new KafkaFutureImpl<>();
@@ -232,12 +236,16 @@ public class MockAdminClient extends AdminClient {
             for (Map.Entry<String, TopicMetadata> topicDescription : allTopics.entrySet())
{
                 String topicName = topicDescription.getKey();
                 if (topicName.equals(requestedTopic) && !topicDescription.getValue().markedForDeletion)
{
-                    TopicMetadata topicMetadata = topicDescription.getValue();
-                    KafkaFutureImpl<TopicDescription> future = new KafkaFutureImpl<>();
-                    future.complete(new TopicDescription(topicName, topicMetadata.isInternalTopic,
topicMetadata.partitions,
-                            Collections.emptySet()));
-                    topicDescriptions.put(topicName, future);
-                    break;
+                    if (topicDescription.getValue().fetchesRemainingUntilVisible > 0)
{
+                        topicDescription.getValue().fetchesRemainingUntilVisible--;
+                    } else {
+                        TopicMetadata topicMetadata = topicDescription.getValue();
+                        KafkaFutureImpl<TopicDescription> future = new KafkaFutureImpl<>();
+                        future.complete(new TopicDescription(topicName, topicMetadata.isInternalTopic,
topicMetadata.partitions,
+                                Collections.emptySet()));
+                        topicDescriptions.put(topicName, future);
+                        break;
+                    }
                 }
             }
             if (!topicDescriptions.containsKey(requestedTopic)) {
@@ -420,6 +428,7 @@ public class MockAdminClient extends AdminClient {
         final boolean isInternalTopic;
         final List<TopicPartitionInfo> partitions;
         final Map<String, String> configs;
+        int fetchesRemainingUntilVisible;
 
         public boolean markedForDeletion;
 
@@ -430,6 +439,7 @@ public class MockAdminClient extends AdminClient {
             this.partitions = partitions;
             this.configs = configs != null ? configs : Collections.emptyMap();
             this.markedForDeletion = false;
+            this.fetchesRemainingUntilVisible = 0;
         }
     }
 
@@ -441,4 +451,12 @@ public class MockAdminClient extends AdminClient {
     public Map<MetricName, ? extends Metric> metrics() {
         return mockMetrics;
     }
+
+    public void setFetchesRemainingUntilVisible(String topicName, int fetchesRemainingUntilVisible)
{
+        TopicMetadata metadata = allTopics.get(topicName);
+        if (metadata == null) {
+            throw new RuntimeException("No such topic as " + topicName);
+        }
+        metadata.fetchesRemainingUntilVisible = fetchesRemainingUntilVisible;
+    }
 }
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 adce304..cb765cc 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
@@ -248,7 +248,7 @@ public final class WorkerUtils {
      * @throws RuntimeException  If one or more topics have different number of partitions
than
      * described in 'topicsInfo'
      */
-    private static void verifyTopics(
+    static void verifyTopics(
         Logger log, AdminClient adminClient,
         Collection<String> topicsToVerify, Map<String, NewTopic> topicsInfo,
int retryCount, long retryBackoffMs) throws Throwable {
 
@@ -279,9 +279,13 @@ public final class WorkerUtils {
                 DescribeTopicsResult topicsResult = adminClient.describeTopics(
                         topicsToVerify, new DescribeTopicsOptions().timeoutMs(ADMIN_REQUEST_TIMEOUT));
                 return topicsResult.all().get();
-            } catch (UnknownTopicOrPartitionException exception) {
-                lastException = exception;
-                Thread.sleep(retryBackoffMs);
+            } catch (ExecutionException exception) {
+                if (exception.getCause() instanceof UnknownTopicOrPartitionException) {
+                    lastException = (UnknownTopicOrPartitionException) exception.getCause();
+                    Thread.sleep(retryBackoffMs);
+                } else {
+                    throw exception;
+                }
             }
         }
         throw lastException;
diff --git a/tools/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java b/tools/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java
index a35efe1..275589b 100644
--- a/tools/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java
+++ b/tools/src/test/java/org/apache/kafka/trogdor/common/WorkerUtilsTest.java
@@ -17,24 +17,23 @@
 
 package org.apache.kafka.trogdor.common;
 
+import static org.junit.Assert.assertEquals;
 
+import org.apache.kafka.clients.admin.MockAdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.TopicPartitionInfo;
-
-import org.apache.kafka.common.Node;
-import org.apache.kafka.clients.admin.MockAdminClient;
-
 import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.apache.kafka.common.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.kafka.clients.admin.NewTopic;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
-import static org.junit.Assert.assertEquals;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -46,7 +45,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
-
 public class WorkerUtilsTest {
 
     private static final Logger log = LoggerFactory.getLogger(WorkerUtilsTest.class);
@@ -318,4 +316,21 @@ public class WorkerUtilsTest {
             tpInfo,
             null);
     }
+
+    @Test
+    public void testVerifyTopics() throws Throwable {
+        Map<String, NewTopic> newTopics = Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC);
+        WorkerUtils.createTopics(log, adminClient, newTopics, true);
+        adminClient.setFetchesRemainingUntilVisible(TEST_TOPIC, 2);
+        WorkerUtils.verifyTopics(log, adminClient, Collections.singleton(TEST_TOPIC),
+            Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), 3, 1);
+        adminClient.setFetchesRemainingUntilVisible(TEST_TOPIC, 100);
+        try {
+            WorkerUtils.verifyTopics(log, adminClient, Collections.singleton(TEST_TOPIC),
+                    Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), 2, 1);
+            Assert.fail("expected to get UnknownTopicOrPartitionException");
+        } catch (UnknownTopicOrPartitionException e) {
+            // expected
+        }
+    }
 }


Mime
View raw message