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

Code before:
      assertFalse(group.subtopologies().isEmpty)

      // Verify the topology contains the expected source and sink topics
      val subtopologies = group.subtopologies().asScala
      assertTrue(subtopologies.exists(subtopology =>
        subtopology.sourceTopics().contains(testTopicName)))

    } finally {
      Utils.closeQuietly(streams, "streams")
      Utils.closeQuietly(client, "adminClient")
    }
  }

  @Test
  def testDescribeStreamsGroupsForStatelessTopology(): Unit = {
    val streamsGroupId = "stream_group_id"
    val testTopicName = "test_topic"
    val testNumPartitions = 1

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

    prepareTopics(List(testTopicName), testNumPartitions)
    prepareRecords(testTopicName)

    val streams = createStreamsGroup(
      inputTopics = Set(testTopicName),
      streamsGroupId = streamsGroupId
    )
    streams.poll(JDuration.ofMillis(500L))

    try {
      TestUtils.waitUntilTrue(() => {
        val firstGroup = client.listGroups().all().get().stream().findFirst().orElse(null)
        firstGroup.groupState().orElse(null) == GroupState.STABLE && firstGroup.groupId() == streamsGroupId
      }, "Streams group did not transition to STABLE before timeout")

      // Verify the describe call works correctly
      val describedGroups = client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
      val group = describedGroups.get(streamsGroupId)
      assertNotNull(group)
      assertEquals(streamsGroupId, group.groupId())
      assertFalse(group.members().isEmpty)
      assertNotNull(group.subtopologies())
      assertFalse(group.subtopologies().isEmpty)

      // Verify the topology contains the expected source and sink topics
      val subtopologies = group.subtopologies().asScala
      assertTrue(subtopologies.exists(subtopology =>
        subtopology.sourceTopics().contains(testTopicName)))


Review comment (bug):
Given the ticket description, should we not check if `assertEquals(GroupState.STABLE, group.groupState())` ? -- Or course, the above "wait" condition makes is somewhat obsolete, but as mentioned above, does this test give us any actual signal that the issue if fixed atm?

If we remove the "wait" checking for STABLE should always succeed with the fix, but fail w/o it (at least occasionally, given it's a race condition)?

Code after:
  def testDescribeStreamsGroupsForStatelessTopology(): Unit = {
    val streamsGroupId = "stream_group_id"
    val testTopicName = "test_topic"
    val testNumPartitions = 1

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

    prepareTopics(List(testTopicName), testNumPartitions)
    prepareRecords(testTopicName)

    val streams = createStreamsGroup(
      inputTopics = Set(testTopicName),
      streamsGroupId = streamsGroupId
    )
    streams.poll(JDuration.ofMillis(500L))

    try {
      TestUtils.waitUntilTrue(() => {
        val firstGroup = client.listGroups().all().get().stream().findFirst().orElse(null)
        firstGroup.groupState().orElse(null) == GroupState.STABLE && firstGroup.groupId() == streamsGroupId
      }, "Streams group did not transition to STABLE before timeout")

      // Verify the describe call works correctly
      val describedGroups = client.describeStreamsGroups(util.List.of(streamsGroupId)).all().get()
      val group = describedGroups.get(streamsGroupId)
      assertNotNull(group)
      assertEquals(streamsGroupId, group.groupId())
      assertFalse(group.members().isEmpty)
      assertNotNull(group.subtopologies())
      assertFalse(group.subtopologies().isEmpty)

      // Verify the topology contains the expected source and sink topics
      val subtopologies = group.subtopologies().asScala
      assertTrue(subtopologies.exists(subtopology =>
        subtopology.sourceTopics().contains(testTopicName)))

      // Test describing a non-existing group
      val nonExistingGroup = "non_existing_stream_group"
      val describedNonExistingGroupResponse = client.describeStreamsGroups(util.List.of(nonExistingGroup))
      assertFutureThrows(classOf[GroupIdNotFoundException], describedNonExistingGroupResponse.all())

    } finally {
      Utils.closeQuietly(streams, "streams")
      Utils.closeQuietly(client, "adminClient")
    }
  }
  
  @Test
  def testDeleteStreamsGroups(): Unit = {
    val testTopicName = "test_topic"