kafka-4295: ConsoleConsumer does not delete the temporary group in zookeeper by huxihx · Pull Request #2054 · apache/kafka

huxihx · 2016-10-22T12:44:45Z

Since consumer stop logic and zk node removal code are in separate threads, so when two threads execute in an interleaving manner, persistent node '/consumers/' might not be removed for those console consumer groups which do not specify "group.id". This will pollute Zookeeper with lots of inactive console consumer offset information.

…roup in zookeeper Author: huxi Since consumer stop logic and zk node removal code are in separate threads, so when two threads execute in an interleaving manner, persistent node '/consumers/<consumer-group>' might not be removed for those console consumer groups which do not specify "group.id". This will pollute Zookeeper with lots of inactive console consumer offset information.

huxihx · 2016-10-24T06:40:17Z

@mjsax seems the failure is not relevant to the commit, how should I handle this situation? Please advice. thanks.

guozhangwang · 2016-10-25T03:52:34Z

@amethystic Could you re-open the PR so that a Jenkins build can be triggered again? cc @hachikuji @ijuma for reviews.

huxihx · 2016-10-26T12:50:47Z

@hachikuji @ijuma please review this PR. Thanks.

hachikuji · 2016-10-26T21:51:46Z

core/src/main/scala/kafka/tools/ConsoleConsumer.scala

-
+        // if we generated a random group id (as none specified explicitly) then avoid polluting zookeeper with persistent group data, this is a hack
+        if (!conf.groupIdPassed && conf.options.has(conf.zkConnectOpt))
+          ZkUtils.maybeDeletePath(conf.options.valueOf(conf.zkConnectOpt), "/consumers/" + conf.consumerProps.get("group.id"))


I think we have a utility in AdminUtils for this.

@hachikuji Yes, although AdminUtils already offers 'deleteConsumerGroupInZK' and 'deleteConsumerGroupInfoForTopicInZK' to implement this, I notice that ConsoleConsumer originally employ this snippet of code to delete inactive console consumer group.
The key point here is we must add this kind of code in addShutdownHook otherwise this thread might not delete the unused zk nodes if the first auto-commit task did not get started.

Makes sense. It looks like Adminutils.deleteConsumerGroupInZK is the one we want. Perhaps we could replace both usages? Actually if we've added this to the shutdown hook, do we still need it in the finally?

The finally block already contains this cleanup call:
consumer.cleanup()
conf.formatter.close()
reportRecordCount()
// if we generated a random group id (as none specified explicitly) then avoid polluting zookeeper with persistent group data, this is a hack if (!conf.groupIdPassed) ZkUtils.maybeDeletePath(conf.options.valueOf(conf.zkConnectOpt), "/consumers/" + conf.consumerProps.get("group.id")) shutdownLatch.countDown()

ZkUtils.maybeDeletePath is idempotent, which can be called many times without any negative impact. I call it from the shutdown hook thread to make sure this thread will try to delete the zkNode after the main thread fails to do this.

As for the question whether we should use deleteConsumerGroupInZK to replace ZkUtils.maybeDeletePath, I recommend we use the original one since only adding two lines in the shutdown hook thread is enough to fix the bug, we do not have to do any other regression test to make sure the new deleteConsumerGroupInZK behaves as expected. But if you insist, I will close this PR and commit a new one with AdminUtils.deleteConsumerGroupInZK.
@hachikuji , what do you think?

As far as I can tell, both methods delegate to ZkClient.deleteRecursive, and they both handle ZkNoNodeException, so I'm not sure I see the concern about idempotence. One difference is that maybeDeletePath catches a Throwable, but I don't see a good reason why we need to preserve that, especially since we're shutting down. Maybe I'm missing something?

Another question I have is about the shutdownLatch.await() below. If we're hitting a shutdown path where the maybeDeletePath is not being executed, wouldn't that mean we end up blocking in await()? Can you clarify the specific case that you're trying to handle in this patch?

Also, you don't need to close the PR if you want to change something. Just push a new commit. Your commits will get squashed when we merge anyway.

The reason why zk node failed to be removed is not that 'maybeDeletePath' is not executed, but because delete(path) got failed in deleteRecursive where parent znode got failed to be deleted if there still existed some children znode before calling delete(path). So if ConsoleConsumer ran exactly long enough to let autocommit thread creates the znode /consumers/<consumer_group>/offsets, everything is good. But if not, when shutting down the consumer, it commits the offset in the interval between deleting children znodes(namely, ids and owners) and deleting the parent znode (/consumers/***). Do I make myself clear?

Besides, I have already tested the patch, with such scenarios:

new consumer case: have to make sure there is no impact for the new consumer

old short-running consumer: successfully deleted

old long-running consumer: successfully deleted

old short/long-running consumer with shut down by 'kill -9': Same behaviour as before, since if JVM process is shut down in this manner, event JVM shut down hook thread cannot be ensured to have a chance to run.

old consumer with groupid specified: have no impact since we only want to clear up those inactive groups without setting groupid

@hachikuji Does it make sense?

@hachikuji any feedbacks about this PR?

Just one question: do we need the additional call to maybeDeletePath in the finally block anymore? Adding it to the shutdown hook seems sufficient based on some testing. Is there any advantage to having it in both locations?

And just one note: it is actually possible to shutdown the console consumer before the shutdown hook gets registered, in which case you can still end up with the leftover node in Zk. You can try trigger this by shutting down the console consumer just after you see it register in zookeeper. This edge case is pretty tricky to address, so it's probably more important to get the main paths.

huxihx · 2016-11-04T00:30:20Z

@joestein Could you help review this PR? Thanks.

huxihx · 2016-11-25T01:41:34Z

@hachikuji Could you address the comments above? Thanks.

hachikuji · 2016-11-30T19:50:55Z

@amethystic Apologies for the delay. I'll get to it today or tomorrow.

huxihx · 2016-12-13T08:29:07Z

@hachikuji Any chances to address the comments these days? Please review the pull request. Thank you!

asfbot · 2016-12-13T09:10:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/97/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2016-12-13T09:10:16Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/96/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2016-12-13T09:16:16Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/95/
Test PASSed (JDK 7 and Scala 2.10).

…' of https://github.com/amethystic/kafka into kafka-4295_ConsoleConsumer_fail_to_remove_zknode_onexit

…Consumer_fail_to_remove_zknode_onexit

1. Remove same cleanup code from within the JVM shutdown hook code block 2. Refine ZkUtils.maybeDeletePath to capture ZkException if to-be-delete znode is not empty

…okeeper remove useless imports in ZkUtils.scala

huxihx · 2016-12-22T07:44:04Z

@hachikuji yes, you are right. As per your comments, I removed some unnecessary code for resource cleanup. ZkUtils.maybeDelete was also refined to capture the ZkException when the to-be-delete path is not empty. The fix only applies for the ConsoleConsumer shutdown via sending SIGINT and SIGTERM. For "kill -9", seems we cannot guarantee it works since application cannot capture it. Do all these make senses?

asfbot · 2016-12-22T08:27:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/361/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2016-12-22T09:28:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/362/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2016-12-22T09:31:11Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/363/
Test PASSed (JDK 8 and Scala 2.11).

During Acceptor initialization, if "Address already in use" error is encountered, the shutdown latch in each Processor is never counted down. Consequently, the Kafka server hangs when `Processor.shutdown` is called. Author: huxi <huxi@zhenrongbao.com> Author: amethystic <huxi_2b@hotmail.com> Reviewers: Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk> Closes #2156 from amethystic/kafka-4428_Kafka_noexit_for_port_already_use

Some of the recent changes to `kafka-run-class.sh` have not been applied to `kafka-run-class.bat`. These recent changes include setting proper streams or connect classpaths. So any streams or connect use case that leverages `kafka-run-class.bat` would fail with an error like ``` Error: Could not find or load main class org.apache.kafka.streams.??? ``` Author: Vahid Hashemian <vahidhashemian@us.ibm.com> Reviewers: Ewen Cheslack-Postava <ewen@confluent.io> Closes #2238 from vahidhashemian/minor/sync_up_kafka-run-class.bat

Mx4jLoader.scala should explicitly `return true` if the class is successfully loaded and started, otherwise it will return false even if the class is loaded. Author: Edward Ribeiro <edward.ribeiro@gmail.com> Reviewers: Ewen Cheslack-Postava <ewen@confluent.io> Closes #2295 from eribeiro/mx4jloader-bug

The original Javadoc description for `ConsumerRecord` is slightly confusing in that it can be read in a way such that an object is a key value pair received from Kafka, but (only) consists of the metadata associated with the record. This PR makes it clearer that the metadata is _included_ with the record, and moves the comma so that the phrase "topic name and partition number" in the sentence is more closely associated with the phrase "from which the record is being received". Author: LoneRifle <LoneRifle@users.noreply.github.com> Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io> Closes #2290 from LoneRifle/patch-1

…ted regex This makes it consistent with MirrorMaker with the old consumer. Author: huxi <huxi@zhenrongbao.com> Author: amethystic <huxi_2b@hotmail.com> Reviewers: Vahid Hashemian <vahidhashemian@us.ibm.com>, Ismael Juma <ismael@juma.me.uk> Closes #2072 from amethystic/kafka-4351_Regex_behavior_change_for_new_consumer

…to be removed does not exist Author: Vahid Hashemian <vahidhashemian@us.ibm.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes #2218 from vahidhashemian/KAFKA-4480

huxihx · 2017-01-01T01:55:36Z

@hachikuji please kindly take some time to review. Thanks.

Author: Himani Arora <1himani.arora@gmail.com> Reviewers: Ismael Juma <ismael@juma.me.uk> Closes #2297 from himani1/refactored_code

zookeeper Addressed Ijuma's comments 1. Restored ZkUtils to trunk code 2. Restored ConsoleConsumerTest to trunk code 3. Restored ZkUtils.maybeDeletePath to trunk code 4. Replaced ZkUtils.maybeDeletePath with AdminUtils.deleteConsumerGroupInZK

asfbot · 2017-01-07T04:17:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/602/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-07T04:17:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/601/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-07T04:17:36Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/600/
Test FAILed (JDK 7 and Scala 2.10).

huxihx · 2017-01-07T04:29:43Z

@ijuma Please review the PR again. Thanks.

…to private

asfbot · 2017-01-09T03:55:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/610/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-09T03:55:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/608/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-09T03:56:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/609/
Test FAILed (JDK 8 and Scala 2.12).

… to private[tools]x

asfbot · 2017-01-09T04:01:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/611/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-09T04:01:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/609/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-09T04:01:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/610/
Test FAILed (JDK 8 and Scala 2.12).

huxihx · 2017-01-12T12:26:40Z

@ijuma Please take time to review this PR. Thanks.

asfbot · 2017-01-12T12:29:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/788/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-12T12:30:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/786/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-12T12:31:40Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/786/
Test FAILed (JDK 8 and Scala 2.12).

huxihx · 2017-01-16T01:42:17Z

@ijuma @hachikuji Please take some time to review the PR. Thanks.

huxihx · 2017-01-20T06:15:44Z

@guozhangwang Do you know how to retest this fix since all the checks have failed although it is not known why it failed.

ijuma

Do I understand correctly that this PR now does the following:

Fix issue where we tried to delete a path in ZK for the new consumer.
Use AdminUtils.deleteConsumerGroupInZK instead of ZKUtils.maybeDeletePath

And the issue described in the JIRA remains since it's an edge case for the old consumer (that we intend to deprecate and remove)?

ijuma · 2017-01-24T12:56:40Z

core/src/test/scala/integration/kafka/tools/ConsoleConsumerIntegrationTest.scala

+      assertTrue("Consumer group should be created.", zkUtils.getChildren(ZkUtils.ConsumersPath).head == groupID)
+    } finally {
+      consumer.stop()
+      ConsoleConsumer.deleteZkPathForConsumerGroup(conf.options.valueOf(conf.zkConnectOpt), conf.consumerProps.getProperty("group.id"))


It seems like the only thing this test is checking is that this call works. And since that is just calling AdminUtils, not sure if the benefit is worth it given the cost of starting up Kafka and ZK.

Agreed, but is there any way we could use to simulate sending terminate signals? This issue is caused by the fact ZkUtils.maybeDeletePath failed to delete the whole directory when ConsoleConsumer exits after receiving a INT or TERM signal , so what we can do is to test if the replaced method deleteConsumerGroupInZK works as expected. Any good idea to test this fix?

huxihx · 2017-01-25T01:39:06Z

For your questions:
1: No, new consumer does not use ZK anymore. The fix is only for old consumer. Old consumer without group id specified leaves a lot of "garage" under /consumers due to the fact that ZKUtils.deletePath failed to delete them.
2: Yes. Tests showed that AdminUtils.deleteConsumerGroupInZK was able to clean up all the unused zk paths.

As what I said above, I am not sure if you guys still want to check in any code for fixing old consumer problems, especially when community is planning to remove it recently. It's up to you. If you don't, I am free to close the PR.
Does it make any sense?

huxihx · 2017-02-03T01:14:07Z

@ijuma What's the status for this PR? Do I address all your comments already?

asfbot · 2017-04-24T13:15:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3141/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-04-24T15:15:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3137/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-24T15:15:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3137/
Test FAILed (JDK 7 and Scala 2.10).

ijuma · 2017-08-04T12:57:54Z

@huxihx sorry for the delay. I was going to look at this PR, but it seems like it includes other changes now. Maybe you could simply revive the part where we fix the new consumer not to call ZK unnecessarily?

huxihx closed this Oct 24, 2016

huxihx reopened this Oct 26, 2016

hachikuji reviewed Oct 26, 2016

View reviewed changes

amethystic added 4 commits December 22, 2016 14:03

Merge branch 'kafka-4295_ConsoleConsumer_fail_to_remove_zknode_onexit…

4185558

…' of https://github.com/amethystic/kafka into kafka-4295_ConsoleConsumer_fail_to_remove_zknode_onexit

Merge remote-tracking branch 'upstream/trunk' into kafka-4295_Console…

0591419

…Consumer_fail_to_remove_zknode_onexit

kafka-ConsoleConsumer does not delete the temporary group in zookeeper

d2767f8

1. Remove same cleanup code from within the JVM shutdown hook code block 2. Refine ZkUtils.maybeDeletePath to capture ZkException if to-be-delete znode is not empty

kafka-4295: ConsoleConsumer does not delete the temporary group in zo…

09438bc

…okeeper remove useless imports in ZkUtils.scala

amethystic and others added 6 commits December 28, 2016 17:34

KAFKA-4480: Report an error in 'kafka-configs' command if the config …

fc7141f

…to be removed does not exist Author: Vahid Hashemian <vahidhashemian@us.ibm.com> Reviewers: Guozhang Wang <wangguoz@gmail.com> Closes #2218 from vahidhashemian/KAFKA-4480

MINOR: Scala code readability improvements

a6f7e81

Author: Himani Arora <1himani.arora@gmail.com> Reviewers: Ismael Juma <ismael@juma.me.uk> Closes #2297 from himani1/refactored_code

amethystic added 2 commits January 7, 2017 12:10

tweaked some imports to keep same as trunk

7a9cfb8

change access level for ConsoleConsumer.deleteZkPathForConsumerGroup …

09699a1

…to private

amethystic added 2 commits January 9, 2017 11:58

changed access level for ConsoleConsumer.deleteZkPathForConsumerGroup…

f52cfa0

… to private[tools]x

removed useless import

c4bf1d9

ijuma reviewed Jan 24, 2017

View reviewed changes

huxihx closed this May 31, 2017

efeg pushed a commit to efeg/kafka that referenced this pull request May 29, 2024

Enable info level log after each goal optimzation (apache#2054)

74416e6

Conversation

huxihx commented Oct 22, 2016

Uh oh!

huxihx commented Oct 24, 2016

Uh oh!

guozhangwang commented Oct 25, 2016

Uh oh!

huxihx commented Oct 26, 2016

Uh oh!

hachikuji Oct 26, 2016

Choose a reason for hiding this comment

Uh oh!

huxihx Oct 26, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Oct 27, 2016

Choose a reason for hiding this comment

Uh oh!

huxihx Oct 27, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Oct 28, 2016

Choose a reason for hiding this comment

Uh oh!

huxihx Oct 28, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

huxihx Nov 1, 2016

Choose a reason for hiding this comment

Uh oh!

hachikuji Dec 13, 2016

Choose a reason for hiding this comment

Uh oh!

huxihx commented Nov 4, 2016

Uh oh!

huxihx commented Nov 25, 2016

Uh oh!

hachikuji commented Nov 30, 2016

Uh oh!

huxihx commented Dec 13, 2016

Uh oh!

asfbot commented Dec 13, 2016

Uh oh!

asfbot commented Dec 13, 2016

Uh oh!

asfbot commented Dec 13, 2016

Uh oh!

huxihx commented Dec 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asfbot commented Dec 22, 2016

Uh oh!

asfbot commented Dec 22, 2016

Uh oh!

asfbot commented Dec 22, 2016

Uh oh!

huxihx commented Jan 1, 2017

Uh oh!

asfbot commented Jan 7, 2017

Uh oh!

asfbot commented Jan 7, 2017

Uh oh!

asfbot commented Jan 7, 2017

Uh oh!

huxihx commented Jan 7, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

Uh oh!

asfbot commented Jan 9, 2017

huxihx Oct 26, 2016 •

edited

Loading

huxihx Oct 27, 2016 •

edited

Loading

huxihx Oct 28, 2016 •

edited

Loading

huxihx commented Dec 22, 2016 •

edited

Loading

huxihx commented Jan 20, 2017 •

edited

Loading

huxihx commented Jan 25, 2017 •

edited

Loading