KAFKA-9538; Fix flaky test testResetOffsetsExportImportPlan (apache…

…#6561) This patch adds logic to the test case to ensure that consumer groups are in a valid state prior to attempting offset reset. Reviewers: Jason Gustafson <[email protected]>
jackin853 · Feb 13, 2020 · 46e80db · 46e80db
1 parent 97d2c72
commit 46e80db
Showing 1 changed file with 12 additions and 2 deletions.
diff --git a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala
@@ -414,8 +414,11 @@ class ResetConsumerGroupOffsetTest extends ConsumerGroupCommandTest {
     val cgcArgs = buildArgsForGroups(Seq(group1, group2), "--all-topics", "--to-offset", "2", "--export")
     val consumerGroupCommand = getConsumerGroupService(cgcArgs)
 
-    produceConsumeAndShutdown(topic = topic1, group = group1, totalMessages = 100, numConsumers = 2)
-    produceConsumeAndShutdown(topic = topic2, group = group2, totalMessages = 100, numConsumers = 5)
+    produceConsumeAndShutdown(topic = topic1, group = group1, totalMessages = 100)
+    produceConsumeAndShutdown(topic = topic2, group = group2, totalMessages = 100)
+
+    awaitConsumerGroupInactive(consumerGroupCommand, group1)
+    awaitConsumerGroupInactive(consumerGroupCommand, group2)
 
     val file = File.createTempFile("reset", ".csv")
     file.deleteOnExit()
@@ -487,6 +490,13 @@ class ResetConsumerGroupOffsetTest extends ConsumerGroupCommandTest {
 
   }
 
+  private def awaitConsumerGroupInactive(consumerGroupService: ConsumerGroupService, group: String): Unit = {
+    TestUtils.waitUntilTrue(() => {
+      val state = consumerGroupService.collectGroupState(group).state
+      state == "Empty" || state == "Dead"
+    }, s"Expected that consumer group is inactive. Actual state: ${consumerGroupService.collectGroupState(group).state}")
+  }
+
   private def resetAndAssertOffsets(args: Array[String],
                                     expectedOffset: Long,
                                     dryRun: Boolean = false,