Fix flaky test `testResetOffsetsExportImportPlan`
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
@@ -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,
