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

Code before:
    try {
      for (i <- 1 to testNumStreamsGroup) {
        val streamsGroupId = s"stream_group_id_$i"
        val streamsConfig = new Properties()
        streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

        val streams = createStreamsGroup(
          configOverrides = streamsConfig,
          inputTopic = testTopicName,
          outputTopic = testOutputTopicName,
          streamsGroupId = streamsGroupId,
        )
        streams.poll(JDuration.ofMillis(500L))
        streamsList += ((streamsGroupId, streams))
      }

      TestUtils.waitUntilTrue(() => {
        val groups = client.listGroups().all().get()
        groups.stream()
          .anyMatch(g => g.groupId().startsWith("stream_group_id_")) &&  testNumStreamsGroup == groups.size()
      }, "Streams groups not ready to delete yet")

      // Test deletion of non-empty existing groups
      var deleteStreamsGroupResult = client.deleteStreamsGroups(targetDeletedGroups)
      assertFutureThrows(classOf[GroupNotEmptyException], deleteStreamsGroupResult.all())
      assertEquals(deleteStreamsGroupResult.deletedGroups().size(),2)

      // Stop and clean up the streams for the groups that are going to be deleted
      streamsList
        .filter { case (groupId, _) => targetDeletedGroups.contains(groupId) }
        .foreach { case (_, streams) =>
          streams.close()
        }

      val listTopicResult = client.listTopics()
      assertEquals(3, listTopicResult.names().get().size())

      // Test deletion of emptied existing streams groups
      deleteStreamsGroupResult = client.deleteStreamsGroups(targetDeletedGroups)
      assertEquals(2, deleteStreamsGroupResult.deletedGroups().size())

      // Wait for the deleted groups to be removed
      TestUtils.waitUntilTrue(() => {
        val groupIds = client.listGroups().all().get().asScala.map(_.groupId()).toSet
        targetDeletedGroups.asScala.forall(id => !groupIds.contains(id))
      }, "Deleted groups not yet deleted")

      // Verify that the deleted groups are no longer present
      val remainingGroups = client.listGroups().all().get()
      assertEquals(targetRemainingGroups.size(), remainingGroups.size())
      remainingGroups.stream().forEach(g => {

Review comment (suggestion):
Missing space after comma in method call. Should be `assertEquals(deleteStreamsGroupResult.deletedGroups().size(), 2)` for consistent formatting.
```suggestion
      assertEquals(deleteStreamsGroupResult.deletedGroups().size(), 2)
```

Code after:

      TestUtils.waitUntilTrue(() => {
        val groups = client.listGroups().all().get()
        groups.stream()
          .anyMatch(g => g.groupId().startsWith("stream_group_id_")) &&  testNumStreamsGroup == groups.size()
      }, "Streams groups not ready to delete yet")

      // Test deletion of non-empty existing groups
      var deleteStreamsGroupResult = client.deleteStreamsGroups(targetDeletedGroups)
      assertFutureThrows(classOf[GroupNotEmptyException], deleteStreamsGroupResult.all())
      assertEquals(2, deleteStreamsGroupResult.deletedGroups().size())

      // Stop and clean up the streams for the groups that are going to be deleted
      streamsList
        .filter { case (groupId, _) => targetDeletedGroups.contains(groupId) }
        .foreach { case (_, streams) =>
          streams.close()
        }

      val listTopicResult = client.listTopics()
      assertEquals(2, listTopicResult.names().get().size())

      // Test deletion of emptied existing streams groups
      deleteStreamsGroupResult = client.deleteStreamsGroups(targetDeletedGroups)
      assertEquals(2, deleteStreamsGroupResult.deletedGroups().size())

      // Wait for the deleted groups to be removed
      TestUtils.waitUntilTrue(() => {
        val groupIds = client.listGroups().all().get().asScala.map(_.groupId()).toSet
        targetDeletedGroups.asScala.forall(id => !groupIds.contains(id))
      }, "Deleted groups not yet deleted")

      // Verify that the deleted groups are no longer present
      val remainingGroups = client.listGroups().all().get()
      assertEquals(targetRemainingGroups.size(), remainingGroups.size())
      remainingGroups.stream().forEach(g => {
        assertTrue(targetRemainingGroups.contains(g.groupId()))
      })

      // Test deletion of a non-existing group
      val nonExistingGroup = "non_existing_stream_group"
      val deleteNonExistingGroupResult = client.deleteStreamsGroups(util.List.of(nonExistingGroup))
      assertFutureThrows(classOf[GroupIdNotFoundException], deleteNonExistingGroupResult.all())
      assertEquals(deleteNonExistingGroupResult.deletedGroups().size(), 1)

    } finally{
      streamsList.foreach { case (_, streams) =>
        streams.close()
      }
      Utils.closeQuietly(client, "adminClient")
    }