KAFKA-5862: Remove ZK dependency from Streams reset tool #3927

bbejeck · 2017-09-20T22:09:53Z

No description provided.

bbejeck · 2017-09-20T22:10:28Z

tedyu · 2017-09-20T22:29:27Z

Please fix the following:

scala2.12/core/src/main/scala/kafka/tools/StreamsResetter.java:20:1: Disallowed import - org.apache.kafka.clients.admin.AdminClient. [ImportControl]
[ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.12/core/src/main/scala/kafka/tools/StreamsResetter.java:21:1: Disallowed import - org.apache.kafka.clients.admin.DeleteTopicsResult. [ImportControl]
[ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.12/core/src/main/scala/kafka/tools/StreamsResetter.java:22:1: Disallowed import - org.apache.kafka.clients.admin.KafkaAdminClient. [ImportControl]

tedyu · 2017-09-20T22:31:21Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+        try {
+            olderAdminClient = kafka.admin.AdminClient.createSimplePlaintext(options.valueOf(bootstrapServerOption));
+            if (!olderAdminClient.describeConsumerGroup(groupId, 0).consumers().get().isEmpty()) {
+                throw new IllegalStateException("Consumer group '" + groupId + "' is still active. "


Include information on the active consumer(s), if possible

tedyu · 2017-09-20T22:32:41Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+            try {
+                entry.getValue().get();
+            } catch (Exception e) {
+                System.err.println("ERROR deleting topic " + entry.getKey() + " failed");


Include exception message

mjsax

Thanks for the patch. Couple of nits.

mjsax · 2017-09-21T04:41:56Z

core/src/main/scala/kafka/tools/StreamsResetter.java

-                30000,
-                30000,
-                JaasUtils.isZkSecurityEnabled());
+            Properties adminClientProperties = new Properties();


Nit: add final

mjsax · 2017-09-21T04:44:22Z

core/src/main/scala/kafka/tools/StreamsResetter.java

        System.out.println("Done.");
    }

+    private void doDelete(List<String> topicsToDelete, KafkaAdminClient adminClient) {


nit: add final -- single parameter per line

mjsax · 2017-09-21T04:44:48Z

core/src/main/scala/kafka/tools/StreamsResetter.java

        System.out.println("Done.");
    }

+    private void doDelete(List<String> topicsToDelete, KafkaAdminClient adminClient) {
+        RuntimeException deleteException = null;
+        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsToDelete);


Nit: final -- also some more below

Same as above: does deleteTopics throw any exceptions? If yes should we handle it?

@guozhangwang yes, but not directly.

Any exception occurring while deleting the topic is thrown with the KafkaFuture.get call for the given topic name from the 'Map<String, KafkaFuture>` map on line 345.

We check for exceptions by iterating over the map on line 347, and any exceptions are logged out, and the first exception is captured and is propagated up to the run method which handles the exception on line 118.

Same for comment below.

mjsax · 2017-09-21T04:50:59Z

streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

+            ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
+            listTopicsOptions.listInternal(true);
+            allTopics.addAll(kafkaAdminClient.listTopics(listTopicsOptions).names().get(30000, TimeUnit.MILLISECONDS));
+            assertThat(allTopics, equalTo(expectedRemainingTopicsAfterCleanup));


Is this safe here? The future that is returned from KafkaAdminClient ensure that the topic was delete? Ie. "delete topic" is not async anymore?

Yes, I believe so. In StreamsResetter we check each returned KafkaFuture for successful completion (lines 347- 354) if it failed for whatever reason the resetter would return with an error return code and fail the test before confirming deletions. Does this answer your question?

Yes. Thanks for confirming.

I was not sure was entry.getValue().get(30, TimeUnit.SECONDS) return -- if I understand you correctly, it either returns "success" or it throws. There is no "failed" return code in the future.

guozhangwang

I have a meta comment about using the new client: currently there is no java doc on the new client about what types of exceptions could possibly be thrown, while from the code it seems certain exceptions are indeed possible. As a user-facing tool we should still consider all the caller functions and see what exceptions may be throwable and whether we should handle it, otherwise it will be thrown all the way to user's face. cc @cmccabe for possible java doc improvements on admin client.

guozhangwang · 2017-09-21T11:31:11Z

core/src/main/scala/kafka/tools/StreamsResetter.java

-            .ofType(String.class)
-            .defaultsTo("localhost:2181")
-            .describedAs("url");
+        zookeeperOption = optionParser.accepts("zookeeper", "Zookeeper option is deprecated");


.. deprecated by bootstrap.servers, as the reset tool would no longer access Zookeeper directly.

guozhangwang · 2017-09-21T11:32:46Z

core/src/main/scala/kafka/tools/StreamsResetter.java

-                throw new IllegalStateException("Consumer group '" + groupId + "' is still active. " +
-                            "Make sure to stop all running application instances before running the reset tool.");
-            }
+            allTopics.addAll(kafkaAdminClient.listTopics().names().get(60, TimeUnit.SECONDS));


Could listTopics throw some TimeoutException? Do we need to handle such cases? cc @cmccabe

The get() method can, but it is handled by line 118.

guozhangwang · 2017-09-21T11:35:23Z

core/src/main/scala/kafka/tools/StreamsResetter.java

-            if (zkUtils != null) {
-                zkUtils.close();
+            if (kafkaAdminClient != null) {
+                kafkaAdminClient.close();


Note that this call is a blocking one, while in the old client it is not.

guozhangwang · 2017-09-21T11:39:06Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+    private void validateNoActiveConsumers(String groupId) {
+        kafka.admin.AdminClient olderAdminClient = null;
+        try {
+            olderAdminClient = kafka.admin.AdminClient.createSimplePlaintext(options.valueOf(bootstrapServerOption));


Could we add a TODO marker to remove this when describeGroup is added to the admin client in case we forget about it?

We should create a JIRA instead of adding a TODO...

added JIRA https://issues.apache.org/jira/browse/KAFKA-5965

guozhangwang · 2017-09-21T11:41:06Z

core/src/main/scala/kafka/tools/StreamsResetter.java

        System.out.println("Done.");
    }

+    private void doDelete(List<String> topicsToDelete, KafkaAdminClient adminClient) {
+        RuntimeException deleteException = null;
+        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsToDelete);


Same as above: does deleteTopics throw any exceptions? If yes should we handle it?

bbejeck · 2017-09-22T19:09:33Z

looks like I messed something up with rebase, let me re-do

EDIT: fixed rebase

bbejeck · 2017-09-22T19:32:23Z

@mjsax @guozhangwang comments addressed.

mjsax

Some more comments.

mjsax · 2017-09-22T20:23:47Z

core/src/main/scala/kafka/tools/StreamsResetter.java

            }
        }

        return exitCode;
    }

+    private void validateNoActiveConsumers(String groupId) {


Nit: add final

mjsax · 2017-09-22T20:24:29Z

core/src/main/scala/kafka/tools/StreamsResetter.java

-            .ofType(String.class)
-            .defaultsTo("localhost:2181")
-            .describedAs("url");
+        zookeeperOption = optionParser.accepts("zookeeper", "Zookeeper option is deprecated by bootstrap.servers, as the reset tool will no longer access Zookeeper directly");


Nit: . missing at the end

mjsax · 2017-09-22T20:25:40Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+        final DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsToDelete);
+        final Map<String, KafkaFuture<Void>> results = deleteTopicsResult.values();
+
+        for (Map.Entry<String, KafkaFuture<Void>> entry : results.entrySet()) {


nit: add final

mjsax · 2017-09-22T20:26:15Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+            try {
+                entry.getValue().get(30, TimeUnit.SECONDS);
+            } catch (Exception e) {
+                System.err.println("ERROR deleting topic " + entry.getKey() + " failed " + e.getMessage());


nit: Error[:]

As we do print all errors here, do we need to include the first detected exception when failing? This implies, that we log this exception twice. Maybe we should just print the whole Stacktrace for each exception, and go a generic error message at the end?

mjsax · 2017-09-22T20:30:53Z

streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

@@ -343,15 +356,13 @@ private void cleanGlobal(final String intermediateUserTopic) {
            parameters = new String[]{
                "--application-id", APP_ID + testNo,
                "--bootstrap-servers", CLUSTER.bootstrapServers(),
-                "--zookeeper", CLUSTER.zKConnectString(),


Maybe we can add one test, that makes sure we don't fail if --zookeeper is specified? -> doNotFailIfZookeeperIsSpecified ?

Good enough to add the --zookeeper args back in? Reason being most existing users will probably keep using the old command.

This should, do. Maybe only add it back for one case so we cover both. And add a comment why we do this.

mjsax · 2017-09-22T20:31:05Z

streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

@@ -374,25 +385,15 @@ private void assertInternalTopicsGotDeleted(final String intermediateUserTopic)
        expectedRemainingTopicsAfterCleanup.add(OUTPUT_TOPIC_2_RERUN);
        expectedRemainingTopicsAfterCleanup.add("__consumer_offsets");

-        Set<String> allTopics;
-        ZkUtils zkUtils = null;
+        Set<String> allTopics = new HashSet<>();


Nit: add final

mjsax · 2017-09-22T20:31:15Z

streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

-            if (zkUtils != null) {
-                zkUtils.close();
-            }
+            ListTopicsOptions listTopicsOptions = new ListTopicsOptions();


mjsax · 2017-09-22T20:31:26Z

streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

+            listTopicsOptions.listInternal(true);
+            allTopics.addAll(kafkaAdminClient.listTopics(listTopicsOptions).names().get(30000, TimeUnit.MILLISECONDS));
+            assertThat(allTopics, equalTo(expectedRemainingTopicsAfterCleanup));
+        } catch (Exception e) {


final

Why do we need to catch an fail here? We can just let the exception bubble out to fail the test

I'd prefer to leave as is because IMHO the Assert.fail is more immediately clear as to why the test failed.

Not sure what @dguy and @guozhangwang think. From my understanding, one should never catch an exception in a test (with the exception you want to do a negative test, ie, the exception implies that the test passes).

Fair enough, I'll take it out.

bbejeck · 2017-09-22T21:20:53Z

@mjsax updates for comments

bbejeck · 2017-09-22T22:44:20Z

@mjsax changes as suggested.

guozhangwang

LGTM. Merged to trunk.

KAFKA-5862: Remove ZK dependency from Streams reset tool

f6c8f2e

tedyu reviewed Sep 20, 2017

View reviewed changes

bbejeck and others added 2 commits September 20, 2017 20:25

KAFKA-5862: fix imports rules

cce547c

KAFKA-5862: fix imports in control file

d1a0239

mjsax requested changes Sep 21, 2017

View reviewed changes

guozhangwang reviewed Sep 21, 2017

View reviewed changes

bbejeck added 4 commits September 22, 2017 12:09

KAFKA-5862: Remove ZK dependency from Streams reset tool

b1cb7e9

KAFKA-5862: fix imports rules

67746db

resolve conflicts

c53374a

KAFKA-5862: updates for comments, rebased, added some timeouts

382f1f2

bbejeck force-pushed the KAFKA-5862_remove_zk_dependency_from_streams_reset_tool branch from 382f1f2 to 6c3ecc5 Compare September 22, 2017 19:30

mjsax requested changes Sep 22, 2017

View reviewed changes

KAFKA-5862: updates for comments

12ac35c

bbejeck force-pushed the KAFKA-5862_remove_zk_dependency_from_streams_reset_tool branch from 6c3ecc5 to 12ac35c Compare September 22, 2017 21:19

KAFKA-5862: keep just one zookeeper arg, don't catch exception in test

72b47e5

asfgit closed this in 271f6b5 Sep 23, 2017

guozhangwang reviewed Sep 23, 2017

View reviewed changes

bbejeck deleted the KAFKA-5862_remove_zk_dependency_from_streams_reset_tool branch July 10, 2024 12:55

KAFKA-5862: Remove ZK dependency from Streams reset tool #3927

KAFKA-5862: Remove ZK dependency from Streams reset tool #3927

Conversation

bbejeck commented Sep 20, 2017

bbejeck commented Sep 20, 2017

tedyu commented Sep 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck Sep 22, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Sep 22, 2017 • edited Loading

bbejeck commented Sep 22, 2017

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck Sep 22, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Sep 22, 2017

bbejeck commented Sep 22, 2017

guozhangwang left a comment

Choose a reason for hiding this comment

bbejeck Sep 22, 2017 •

edited

Loading

guozhangwang left a comment •

edited

Loading

bbejeck commented Sep 22, 2017 •

edited

Loading

bbejeck Sep 22, 2017 •

edited

Loading