Repository: apache/kafka
File: core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala (Scala)

Code before:
      } finally {
        backgroundConsumers.close()
      }
    } finally {
      Utils.closeQuietly(client, "adminClient")
    }
  }

  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
  @MethodSource(Array("getTestGroupProtocolParametersAll"))
  def testDescribeGroups(groupProtocol: String): Unit = {
    val config = createConfig
    client = Admin.create(config)
    try {
      assertConsumerGroupsIsClean()

      val testTopicName = "test_topic"
      val testTopicName1 = testTopicName + "1"
      val testTopicName2 = testTopicName + "2"
      val testNumPartitions = 2
      prepareTopics(List(testTopicName, testTopicName1, testTopicName2), testNumPartitions)

      val producer = createProducer()
      try {
        producer.send(new ProducerRecord(testTopicName, 0, null, null)).get()
      } finally {
        Utils.closeQuietly(producer, "producer")
      }

      val testGroupId = "test_group_id"
      val testClientId = "test_client_id"
      val testInstanceId1 = "test_instance_id_1"
      val testInstanceId2 = "test_instance_id_2"
      val fakeGroupId = "fake_group_id"

      // contains two static members and one dynamic member
      val groupInstances = Set(testInstanceId1, testInstanceId2, "")
      val topics = Set(testTopicName, testTopicName1, testTopicName2)

      // We need to disable the auto commit because after the members got removed from group, the offset commit
      // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
      val defaultConsumerConfig = new Properties(consumerConfig)
      defaultConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
      defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
      defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
      val backgroundConsumers = prepareConsumers(groupInstances, topics, defaultConsumerConfig)

      try {
        val groupType = if (groupProtocol.equalsIgnoreCase(GroupProtocol.CONSUMER.name)) GroupType.CONSUMER else GroupType.CLASSIC
        // Start consumer polling threads in the background
        backgroundConsumers.start()

Review comment (suggestion):
I add it to deflaky for assertion of groupEpoch.
Not sure why the empty topic will lead to groupEpoch changing. 
I'm still investigating.

Code after:
      }
    } finally {
      Utils.closeQuietly(client, "adminClient")
    }
  }

  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedGroupProtocolNames)
  @MethodSource(Array("getTestGroupProtocolParametersAll"))
  def testDescribeGroups(groupProtocol: String): Unit = {
    val config = createConfig
    client = Admin.create(config)
    try {
      assertConsumerGroupsIsClean()

      val testTopicName = "test_topic"
      val testTopicName1 = testTopicName + "1"
      val testTopicName2 = testTopicName + "2"
      val testNumPartitions = 2
      prepareTopics(List(testTopicName, testTopicName1, testTopicName2), testNumPartitions)

      val producer = createProducer()
      try {
        producer.send(new ProducerRecord(testTopicName, 0, null, null)).get()
      } finally {
        Utils.closeQuietly(producer, "producer")
      }

      val testGroupId = "test_group_id"
      val testClientId = "test_client_id"
      val testInstanceId1 = "test_instance_id_1"
      val testInstanceId2 = "test_instance_id_2"
      val fakeGroupId = "fake_group_id"

      // contains two static members and one dynamic member
      val groupInstances = Set(testInstanceId1, testInstanceId2, "")
      val topics = Set(testTopicName, testTopicName1, testTopicName2)

      // We need to disable the auto commit because after the members got removed from group, the offset commit
      // will cause the member rejoining and the test will be flaky (check ConsumerCoordinator#OffsetCommitResponseHandler)
      val defaultConsumerConfig = new Properties(consumerConfig)
      defaultConsumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
      defaultConsumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, testGroupId)
      defaultConsumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, testClientId)
      val backgroundConsumers = prepareConsumers(groupInstances, topics, defaultConsumerConfig)

      try {
        val groupType = if (groupProtocol.equalsIgnoreCase(GroupProtocol.CONSUMER.name)) GroupType.CONSUMER else GroupType.CLASSIC
        // Start consumer polling threads in the background
        backgroundConsumers.start()

        val describeWithFakeGroupResult = client.describeConsumerGroups(util.List.of(testGroupId, fakeGroupId),