KAFKA-7938: Fix test flakiness in DeleteConsumerGroupsTest (#6312)

core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala

@@ -70,11 +70,11 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
     val service = getConsumerGroupService(cgcArgs)
 
     TestUtils.waitUntilTrue(() => {
-      service.listGroups().contains(group)
+      service.collectGroupMembers(false)._2.get.size == 1
     }, "The group did not initialize as expected.", maxRetries = 3)
 
     val output = TestUtils.grabConsoleOutput(service.deleteGroups())
-    assertTrue(s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group",
+    assertTrue(s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group. Output was: (${output})",
       output.contains(s"Group '$group' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message))
   }
 
@@ -88,11 +88,12 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
     val service = getConsumerGroupService(cgcArgs)
 
     TestUtils.waitUntilTrue(() => {
-      service.listGroups().contains(group)
+      service.collectGroupMembers(false)._2.get.size == 1
     }, "The group did not initialize as expected.", maxRetries = 3)
 
     val result = service.deleteGroups()
-    assertTrue(s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group",
+    assertNotNull(s"Group was deleted successfully, but it shouldn't have been. Result was:(${result})", result.get(group).get)
+    assertTrue(s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group. Result was:(${result})",
       result.size == 1 && result.keySet.contains(group) && result.get(group).get.getCause.isInstanceOf[GroupNotEmptyException])
   }
 
@@ -199,29 +200,6 @@ class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
         result.keySet.contains(missingGroup) && result.get(missingGroup).get.getMessage.contains(Errors.GROUP_ID_NOT_FOUND.message))
   }
 
-  @Test
-  def testDeleteCmdWithShortInitialization() {
-    // run one consumer in the group
-    addConsumerGroupExecutor(numConsumers = 1)
-    val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group)
-    val service = getConsumerGroupService(cgcArgs)
-
-    val output = TestUtils.grabConsoleOutput(service.deleteGroups())
-    assertTrue(s"The consumer group deletion did not work as expected",
-      output.contains(s"Group '$group' could not be deleted due to"))
-  }
-
-  @Test
-  def testDeleteWithShortInitialization() {
-    // run one consumer in the group
-    addConsumerGroupExecutor(numConsumers = 1)
-    val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group)
-    val service = getConsumerGroupService(cgcArgs)
-
-    val result = service.deleteGroups()
-    assertTrue(s"The consumer group deletion did not work as expected",
-      result.size == 1 && result.keySet.contains(group))
-  }
 
   @Test(expected = classOf[OptionException])
   def testDeleteWithUnrecognizedNewConsumerOption() {