You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2020/03/09 13:36:51 UTC
[kafka] branch 2.3 updated: KAFKA-8245: Fix Flaky Test
DeleteConsumerGroupsTest#testDeleteCmdAllGroups (#8032) (#8241)
This is an automated email from the ASF dual-hosted git repository.
ijuma pushed a commit to branch 2.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.3 by this push:
new 002f732 KAFKA-8245: Fix Flaky Test DeleteConsumerGroupsTest#testDeleteCmdAllGroups (#8032) (#8241)
002f732 is described below
commit 002f732244a57837fd15d223c7ff39e8bf661475
Author: Vikas Singh <vi...@confluent.io>
AuthorDate: Mon Mar 9 06:35:48 2020 -0700
KAFKA-8245: Fix Flaky Test DeleteConsumerGroupsTest#testDeleteCmdAllGroups (#8032) (#8241)
Change unit tests to make sure the consumer group is in Stable state (i.e. consumers have completed joining the group)
(cherry picked from commit 350dce865ae5420a25496bc502c55de4c15bf71e)
Reviewers: Ismael Juma <is...@juma.me.uk>
Co-authored-by: Chia-Ping Tsai <ch...@gmail.com>
---
.../scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala | 11 ++++++-----
1 file changed, 6 insertions(+), 5 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala
index 29c2915..5a81aa4 100644
--- a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala
@@ -107,7 +107,7 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
val service = getConsumerGroupService(cgcArgs)
TestUtils.waitUntilTrue(() => {
- service.listGroups().contains(group)
+ service.listGroups().contains(group) && service.collectGroupState(group).state == "Stable"
}, "The group did not initialize as expected.", maxRetries = 3)
executor.shutdown()
@@ -137,7 +137,8 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
val service = getConsumerGroupService(cgcArgs)
TestUtils.waitUntilTrue(() => {
- service.listGroups().forall(groupId => groups.keySet.contains(groupId))
+ service.listGroups().toSet == groups.keySet &&
+ groups.keySet.forall(groupId => service.collectGroupState(groupId).state == "Stable")
}, "The group did not initialize as expected.", maxRetries = 3)
// Shutdown consumers to empty out groups
@@ -168,7 +169,7 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
val service = getConsumerGroupService(cgcArgs)
TestUtils.waitUntilTrue(() => {
- service.listGroups().contains(group)
+ service.listGroups().contains(group) && service.collectGroupState(group).state == "Stable"
}, "The group did not initialize as expected.", maxRetries = 3)
executor.shutdown()
@@ -193,7 +194,7 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
val service = getConsumerGroupService(cgcArgs)
TestUtils.waitUntilTrue(() => {
- service.listGroups().contains(group)
+ service.listGroups().contains(group) && service.collectGroupState(group).state == "Stable"
}, "The group did not initialize as expected.", maxRetries = 3)
executor.shutdown()
@@ -220,7 +221,7 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
val service = getConsumerGroupService(cgcArgs)
TestUtils.waitUntilTrue(() => {
- service.listGroups().contains(group)
+ service.listGroups().contains(group) && service.collectGroupState(group).state == "Stable"
}, "The group did not initialize as expected.", maxRetries = 3)
executor.shutdown()