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

Code before:
    val classicGroupId = "classic_group_id"
    val consumerGroupId = "consumer_group_id"
    val shareGroupId = "share_group_id"
    val simpleGroupId = "simple_group_id"
    val testTopicName = "test_topic"

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
    val classicGroupConfig = new Properties(consumerConfig)
    classicGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, classicGroupId)
    val classicGroup = createConsumer(configOverrides = classicGroupConfig)

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
    val consumerGroupConfig = new Properties(consumerConfig)
    consumerGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId)
    val consumerGroup = createConsumer(configOverrides = consumerGroupConfig)

    val shareGroupConfig = new Properties(consumerConfig)
    shareGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, shareGroupId)
    val shareGroup = createShareConsumer(configOverrides = shareGroupConfig)

    val config = createConfig
    client = Admin.create(config)
    try {
      client.createTopics(util.Set.of(
        new NewTopic(testTopicName, 1, 1.toShort)
      )).all().get()
      waitForTopics(client, List(testTopicName), List())
      val topicPartition = new TopicPartition(testTopicName, 0)

      classicGroup.subscribe(util.Set.of(testTopicName))
      classicGroup.poll(JDuration.ofMillis(1000))
      consumerGroup.subscribe(util.Set.of(testTopicName))
      consumerGroup.poll(JDuration.ofMillis(1000))
      shareGroup.subscribe(util.Set.of(testTopicName))
      shareGroup.poll(JDuration.ofMillis(1000))

      val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(simpleGroupId,
        util.Map.of(topicPartition, new OffsetAndMetadata(0L)))
      assertNull(alterConsumerGroupOffsetsResult.all().get())
      assertNull(alterConsumerGroupOffsetsResult.partitionResult(topicPartition).get())

      TestUtils.waitUntilTrue(() => {
        val groups = client.listGroups().all().get()
        groups.size() == 4
      }, "Expected to find all groups")

      val classicGroupListing = new GroupListing(classicGroupId, Optional.of(GroupType.CLASSIC), "consumer", Optional.of(GroupState.STABLE))
      val consumerGroupListing = new GroupListing(consumerGroupId, Optional.of(GroupType.CONSUMER), "consumer", Optional.of(GroupState.STABLE))
      val shareGroupListing = new GroupListing(shareGroupId, Optional.of(GroupType.SHARE), "share", Optional.of(GroupState.STABLE))
      val simpleGroupListing = new GroupListing(simpleGroupId, Optional.of(GroupType.CLASSIC), "", Optional.of(GroupState.EMPTY))


Review comment (bug):
The consumer may create `testTopicName` due to a metadata update, causing the subsequent `client.createTopics` call to fail since the topic already exists.

see https://github.com/apache/kafka/actions/runs/17495296037/job/49695550299?pr=20472

@lucliu1108  WDYT? Perhaps we could create the topic first?

Code after:
  def testListGroups(): Unit = {
    val classicGroupId = "classic_group_id"
    val consumerGroupId = "consumer_group_id"
    val shareGroupId = "share_group_id"
    val simpleGroupId = "simple_group_id"
    val streamsGroupId = "streams_group_id"
    val testTopicName = "test_topic"

    val config = createConfig
    client = Admin.create(config)

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name)
    val classicGroupConfig = new Properties(consumerConfig)
    classicGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, classicGroupId)
    val classicGroup = createConsumer(configOverrides = classicGroupConfig)

    consumerConfig.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.CONSUMER.name)
    val consumerGroupConfig = new Properties(consumerConfig)
    consumerGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId)
    val consumerGroup = createConsumer(configOverrides = consumerGroupConfig)

    val shareGroupConfig = new Properties(consumerConfig)
    shareGroupConfig.put(ConsumerConfig.GROUP_ID_CONFIG, shareGroupId)
    val shareGroup = createShareConsumer(configOverrides = shareGroupConfig)

    val streamsGroup = createStreamsGroup(
      inputTopic = testTopicName,
      streamsGroupId = streamsGroupId
    )

    try {
      client.createTopics(util.Set.of(
        new NewTopic(testTopicName, 1, 1.toShort)
      )).all().get()
      waitForTopics(client, List(testTopicName), List())
      val topicPartition = new TopicPartition(testTopicName, 0)

      classicGroup.subscribe(util.Set.of(testTopicName))
      classicGroup.poll(JDuration.ofMillis(1000))
      consumerGroup.subscribe(util.Set.of(testTopicName))
      consumerGroup.poll(JDuration.ofMillis(1000))
      shareGroup.subscribe(util.Set.of(testTopicName))
      shareGroup.poll(JDuration.ofMillis(1000))
      streamsGroup.poll(JDuration.ofMillis(1000))

      val alterConsumerGroupOffsetsResult = client.alterConsumerGroupOffsets(simpleGroupId,
        util.Map.of(topicPartition, new OffsetAndMetadata(0L)))
      assertNull(alterConsumerGroupOffsetsResult.all().get())
      assertNull(alterConsumerGroupOffsetsResult.partitionResult(topicPartition).get())

      TestUtils.waitUntilTrue(() => {