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

Code before:

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

      // Verify that there are 3 groups created
      val groups = client.listGroups().all().get()
      assertEquals(testNumStreamsGroup, groups.size())

      // 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(java.time.Duration.ofSeconds(10))
          streams.cleanUp()
        }

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

      // 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(java.time.Duration.ofSeconds(10))
        streams.cleanUp()
      }

Review comment (suggestion):
Could we also list topics and check whether deleting groups impacted deleting topics as well? Do we have internal topics as well? Would be nice to check them as well.

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

object PlaintextAdminIntegrationTest {