MINOR: Remove deprecated ZkUtils usage from EmbeddedKafkaCluster #5324

omkreddy · 2018-07-03T16:54:36Z

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

omkreddy · 2018-07-03T16:55:04Z

@mjsax Please take a look at this minor change.

ijuma · 2018-07-05T15:54:42Z

Can we use AdminClient instead?

omkreddy · 2018-07-05T16:10:30Z

Yes, We can use AdminClient. Currently Streams tests uses ZKUtils/KafkaZkClient/AdminZkClient classes. If there are no concerns, I can migrate these calls to AdminClient.

ijuma · 2018-07-05T16:18:26Z

Yes, I think it would be best to avoid using KafkaZkClient outside of core. We should update the checkstyle rules to enforce this once usages have been migrated.

omkreddy · 2018-07-06T02:30:09Z

retest this please

omkreddy · 2018-07-06T18:29:07Z

@ijuma @mjsax Updated the PR. Please take a look when you get a chance.

omkreddy · 2018-07-12T15:36:47Z

@mjsax @guozhangwang Call for review.

guozhangwang · 2018-07-16T04:47:02Z

cc @cmccabe @rajinisivaram

mjsax

Thanks for the PR. Couple of comments. Mostly nits.

mjsax · 2018-07-16T20:47:40Z

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

-            for (Map.Entry<String, Properties> topicConfig : topicConfigs.entrySet()) {
-                if (topicConfig.getKey().equals(changelog)) {
-                    return topicConfig.getValue();
+        try (AdminClient adminClient = createAdminClient()) {


nit: add final

We apply final where ever possible in Streams code base -- for all variables, parameters etc. -- I won't comment on the other missing final, please add them thought the whole PR. There are 5 missing final in the next lines for example. Thx.

mjsax · 2018-07-16T20:51:15Z

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

+            try {
+                Config config =  adminClient.describeConfigs(Collections.singletonList(configResource)).values().get(configResource).get();
+                Properties properties = new Properties();
+                for (ConfigEntry configEntry: config.entries()) {


nit: add space configEntry : -- wondering why this is not detected by checkstyle...

mjsax · 2018-07-16T20:52:29Z

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

+                Config config =  adminClient.describeConfigs(Collections.singletonList(configResource)).values().get(configResource).get();
+                Properties properties = new Properties();
+                for (ConfigEntry configEntry: config.entries()) {
+                    if (configEntry.source() ==  ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG)


nit: add { } -- we use curly braces for all blocks

nit: remove double space after ==

mjsax · 2018-07-16T20:53:02Z

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

+        try (AdminClient adminClient = createAdminClient()) {
+            ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, changelog);
+            try {
+                Config config =  adminClient.describeConfigs(Collections.singletonList(configResource)).values().get(configResource).get();


nit: remove double space after =

mjsax · 2018-07-16T21:03:32Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java

+                final Set<String> allTopics = adminClient.listTopics(new ListTopicsOptions().listInternal(true)).names().get();
+                return !allTopics.removeAll(deletedTopics);
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(e);


Is re-throwing the best approach? We could also return false and time-out?

mjsax · 2018-07-16T21:03:39Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java

+                final Set<String> allTopics = adminClient.listTopics(new ListTopicsOptions().listInternal(true)).names().get();
+                return allTopics.equals(remainingTopics);
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(e);


mjsax · 2018-07-16T21:06:47Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

+        newTopic.configs((Map) topicConfig);
+
+        int tries = 0;
+        int maxTries = 5;


Can't we configure AdminClient with internal retries instead?

mjsax · 2018-07-16T21:08:15Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

-    private KafkaZkClient createZkClient() {
-        return KafkaZkClient.apply(zookeeperConnect(), false, DEFAULT_ZK_SESSION_TIMEOUT_MS,
-                DEFAULT_ZK_CONNECTION_TIMEOUT_MS, Integer.MAX_VALUE, Time.SYSTEM, "testMetricGroup", "testMetricType");
+    private AdminClient createAdminClient() {


Seems to be redundant with the method in EmbeddedKafkaCluster.java -- might be better to make this method public and call within EmbeddedKafkaCluster.java ?

omkreddy · 2018-07-18T20:50:06Z

retest this please

mjsax

Thanks for the update. Some more nits.

mjsax · 2018-07-18T21:46:49Z

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

+        try (final AdminClient adminClient = createAdminClient()) {
+            final ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, changelog);
+            try {
+                final Config config =  adminClient.describeConfigs(Collections.singletonList(configResource)).values().get(configResource).get();


nit: double space after =

Similarly, adminClient.describeConfigs does not read from ZK but from broker cache, and hence maybe subject to race conditions.

hmm.. ran multiple times with out any issues. not sure how to handle this? Can we add small Thread.sleep to method? or some kind of waitForCondition implementation?

Thinking about this a bit more, using adminClient.describeConfigs here would be okay since we never dynamically change the config, we just want to test that the configs specified when the topic was created is expected. So if the topic creation is not yet propagated, it will fail and re-try on admin client anyways. So nvm.

mjsax · 2018-07-18T21:47:15Z

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

                }
+                return properties;
+            } catch (InterruptedException | ExecutionException e) {


nit: add final

mjsax · 2018-07-18T21:47:55Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java

+     * @param timeoutMs the max time to wait for the topics to be deleted (does not block if {@code <= 0})
+     */
+    public void deleteAllTopicsAndWait(final long timeoutMs) throws InterruptedException {
+        Set<String> topics = brokers[0].listTopics();


nit: add final

I'm a bit concerned using listTopics than using JavaConverters.seqAsJavaListConverter(brokers[0].kafkaServer().zkClient().getAllTopicsInCluster()).asJava()) as only the latter can get the source-of-truth on ZK while the former may be subject to race conditions (e.g. if you create the topic and then call listTopics, it may not be included if the metadata was not propagated yet).

This is a one-node cluster, though, right?

/**

Runs an in-memory, "embedded" Kafka cluster with 1 ZooKeeper instance and 1 Kafka broker.
*/
public class EmbeddedKafkaCluster extends ExternalResource {

The javadoc is out dated: we actually create a KafkaEmbedded[] brokers.

If there are race conditions, we need to fix them. Remember that Streams itself has to use AdminClient.

@guozhangwang Since I was using deleteAllTopicsAndWait only in AbstractJoinIntegrationTest.cleanup(), So I thought of using listTopics(). But yes, its is error prone in delete-check-create scenarios. I changed the the implementation back to using KafkaZkClient.

mjsax · 2018-07-18T21:48:55Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

+
+        try (final AdminClient adminClient = createAdminClient()) {
+            adminClient.createTopics(Collections.singletonList(newTopic)).all().get();
+        } catch (InterruptedException | ExecutionException e) {


nit: add final

mjsax · 2018-07-18T21:49:22Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

    }

    public void deleteTopic(final String topic) {
        log.debug("Deleting topic { name: {} }", topic);
+        try (final AdminClient adminClient = createAdminClient()) {
+            adminClient.deleteTopics(Collections.singletonList(topic)).all().get();
+        } catch (InterruptedException | ExecutionException e) {


nit: add final

mjsax · 2018-07-18T21:51:05Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

+        try (final AdminClient adminClient = createAdminClient()) {
+            adminClient.deleteTopics(Collections.singletonList(topic)).all().get();
+        } catch (InterruptedException | ExecutionException e) {
+            if (!(e.getCause() instanceof UnknownTopicOrPartitionException))


Might be better to add a catch-clause for this case instead?

try { ... } catch (final UnknownTopicOrPartitionException ignoreAndSwallow) { } catch (final InterruptedException | ExecutionException e) { throw new RuntimeException(e); }

UnknownTopicOrPartitionException is the cause of the actual exception e, so we cannot just catch it here.

Makes sense.

mjsax · 2018-07-18T21:51:18Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

+    public Set<String> listTopics() {
+        try (final AdminClient adminClient = createAdminClient()) {
+            return adminClient.listTopics(new ListTopicsOptions()).names().get();
+        } catch (InterruptedException | ExecutionException e) {


nit: add final

omkreddy · 2018-07-19T07:27:08Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/KafkaEmbedded.java

@@ -93,6 +97,7 @@ private Properties effectiveConfigFrom(final Properties initialConfig) throws IO
        effectiveConfig.put(KafkaConfig$.MODULE$.AutoCreateTopicsEnableProp(), true);
        effectiveConfig.put(KafkaConfig$.MODULE$.MessageMaxBytesProp(), 1000000);
        effectiveConfig.put(KafkaConfig$.MODULE$.ControlledShutdownEnableProp(), true);
+        effectiveConfig.put(KafkaConfig$.MODULE$.ZkSessionTimeoutMsProp(), 10000);


@ijuma @guozhangwang I am observing transient zk session timeout error in tests. So I updated the zk session timeout to 10s. let me know if any concern.

kafka.zookeeper.ZooKeeperClientTimeoutException: Timed out waiting for connection while in state: CONNECTING at kafka.zookeeper.ZooKeeperClient$$anonfun$kafka$zookeeper$ZooKeeperClient$$waitUntilConnected$1.apply$mcV$sp(ZooKeeperClient.scala:225) at kafka.zookeeper.ZooKeeperClient$$anonfun$kafka$zookeeper$ZooKeeperClient$$waitUntilConnected$1.apply(ZooKeeperClient.scala:221) at kafka.zookeeper.ZooKeeperClient$$anonfun$kafka$zookeeper$ZooKeeperClient$$waitUntilConnected$1.apply(ZooKeeperClient.scala:221) at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251) at kafka.zookeeper.ZooKeeperClient.kafka$zookeeper$ZooKeeperClient$$waitUntilConnected(ZooKeeperClient.scala:221) at kafka.zookeeper.ZooKeeperClient.<init>(ZooKeeperClient.scala:95) at kafka.zk.KafkaZkClient$.apply(KafkaZkClient.scala:1580) at kafka.server.KafkaServer.kafka$server$KafkaServer$$createZkClient$1(KafkaServer.scala:348) at kafka.server.KafkaServer.initZkClient(KafkaServer.scala:372) at kafka.server.KafkaServer.startup(KafkaServer.scala:202) at kafka.utils.TestUtils$.createServer(TestUtils.scala:135) at kafka.utils.TestUtils.createServer(TestUtils.scala) at org.apache.kafka.streams.integration.utils.KafkaEmbedded.<init>(KafkaEmbedded.java:77)

I've seen this before as well, but can never observe it locally. @omkreddy are you certain it is due to short timeout value (default is 6 seconds so it is already pretty large, or put it another way increasing to 10 from 6 may not safely avoid if it is really a timeout race condition right?)

omkreddy · 2018-07-19T07:28:12Z

@mjsax @guozhangwang Thanks for the review. Updated the PR.

guozhangwang · 2018-07-19T17:00:58Z

The PR lgtm, other than the comment above. @omkreddy could you investigate a bit more on the root cause of ZK timeout? 6 seconds seems pretty large to me, and if it is indeed still not sufficient, I'm fine with increasing it to even larger number, like 30 seconds.

omkreddy · 2018-07-19T17:37:27Z

@guozhangwang ZK timeout error is transient. My random observation is, Its frequent after recent zk library upgrade. But I am not sure.

Currently, In core tests, we configured 10seconds as timeout, So we can also use 10 seconds here.
I am fine with increasing timeout to 30 seconds.

ijuma · 2018-07-19T18:31:29Z

@omkreddy Does it go away completely after the timeout bump?

omkreddy · 2018-07-19T18:48:43Z

retest this please

omkreddy · 2018-07-19T19:42:25Z

@ijuma yes. I am not able to reproduce with 10sec. I ran few times. But I am getting timeout error with 6seconds. also recent jenkins builds streams tests are failing with timeout error.

As suggested by Guozhang, maybe we can increase timeout to a larger number

guozhangwang · 2018-07-19T21:27:12Z

Thanks @omkreddy , I'm going to merge this PR as is, and will continue investigating the flaky test. If it is still due to the ZK timeout I will try to further bump it up in another PR.

mjsax added the streams label Jul 3, 2018

omkreddy force-pushed the streamstest branch from 3b3859d to 7a5d2fc Compare July 5, 2018 21:45

mjsax reviewed Jul 16, 2018

View reviewed changes

omkreddy added 3 commits July 19, 2018 00:43

MINOR: Remove deprecated ZkUtils usage from EmbeddedKafkaCluster

bae60c9

Use AdminClient in Streams tests

25894b5

Update checkstyle import-control.xml

971cbeb

omkreddy force-pushed the streamstest branch from fe7cc5d to 511be5b Compare July 18, 2018 20:21

Address review comments

e79ee89

omkreddy force-pushed the streamstest branch from 511be5b to e79ee89 Compare July 18, 2018 20:23

mjsax reviewed Jul 18, 2018

View reviewed changes

Address review comments

079bcf9

omkreddy commented Jul 19, 2018

View reviewed changes

mjsax approved these changes Jul 19, 2018

View reviewed changes

guozhangwang merged commit 96c53e9 into apache:trunk Jul 19, 2018

omkreddy deleted the streamstest branch July 20, 2018 03:06

mjsax mentioned this pull request Aug 5, 2018

MINOR: increase AdminClient retries for integration tests #5462

Closed

3 tasks

MINOR: Remove deprecated ZkUtils usage from EmbeddedKafkaCluster #5324

MINOR: Remove deprecated ZkUtils usage from EmbeddedKafkaCluster #5324

Conversation

omkreddy commented Jul 3, 2018

Committer Checklist (excluded from commit message)

omkreddy commented Jul 3, 2018

ijuma commented Jul 5, 2018

omkreddy commented Jul 5, 2018 • edited Loading

ijuma commented Jul 5, 2018

omkreddy commented Jul 6, 2018

omkreddy commented Jul 6, 2018

omkreddy commented Jul 12, 2018

guozhangwang commented Jul 16, 2018

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

Choose a reason for hiding this comment

omkreddy commented Jul 18, 2018

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

omkreddy Jul 19, 2018 • 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

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

omkreddy commented Jul 19, 2018

guozhangwang commented Jul 19, 2018

omkreddy commented Jul 19, 2018 • edited Loading

ijuma commented Jul 19, 2018

omkreddy commented Jul 19, 2018

omkreddy commented Jul 19, 2018 • edited Loading

guozhangwang commented Jul 19, 2018

omkreddy commented Jul 5, 2018 •

edited

Loading

omkreddy Jul 19, 2018 •

edited

Loading

omkreddy commented Jul 19, 2018 •

edited

Loading

omkreddy commented Jul 19, 2018 •

edited

Loading