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

Code before:
    }
  }

  @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
      }, "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 (bug):
Error message: "Stream group not stable yet"
Correct me if I'm wrong, but this is final error message (when timeout is reached), so error message should be different

Code after:

    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"
    val testNumPartitions = 3
    val testNumStreamsGroup = 3

    val targetDeletedGroups = util.List.of("stream_group_id_2", "stream_group_id_3")
    val targetRemainingGroups = util.List.of("stream_group_id_1")

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

    prepareTopics(List(testTopicName), testNumPartitions)