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

Code before:
      }
    }
  }

  @Test
  def testDescribeStreamsGroups(): 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(
      inputTopic = 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
      }, "Stream group not stable yet")

      // 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")
    }

Review comment (suggestion):
This logic assumes the first group in the list is the streams group, but there's no guarantee of ordering. It should filter for the specific `streamsGroupId` instead of using `findFirst()`.
```suggestion
        val streamsGroupOpt = client.listGroups().all().get().stream()
          .filter(g => g.groupId() == streamsGroupId)
          .findFirst()
          .orElse(null)
        streamsGroupOpt != null && streamsGroupOpt.groupState().orElse(null) == GroupState.STABLE
```

Code after:
            case _: InterruptException => // ignore
          } finally {
            consumer.close()
            stopLatch.countDown()
          }
        }
      }
    }
  }

  @Test
  def testDescribeStreamsGroups(): 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(
      inputTopic = testTopicName,
      streamsGroupId = streamsGroupId
    )
    streams.poll(JDuration.ofMillis(500L))

    try {
      TestUtils.waitUntilTrue(() => {
        val firstGroup = client.listGroups().all().get().stream()
          .filter(g => g.groupId() == streamsGroupId).findFirst().orElse(null)
        firstGroup.groupState().orElse(null) == GroupState.STABLE && firstGroup.groupId() == streamsGroupId
      }, "Stream group not stable yet")

      // 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"