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

Code before:
      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)))

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


Review comment (bug):
Does this test really do, what it is supposed to verify?

If we wait here, until state is STABLE, a potential race condition that the state is reported as NOT_READY, even if it is READY seems not to get flagged?

Would we not need to ensure, that the broker side state is READY (not sure if we can verify this), and check if the client state is incorrectly reported as NOT_READY?

Code after:
      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)))

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