Repository: apache/kafka
File: test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/ClusterInstance.java (Java)

Code before:
            // Ensure that logs from all replicas are deleted
            TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
                    topicPartitions.stream().allMatch(tp -> broker.logManager().getLog(tp, false).isEmpty())),
                "Replica logs not deleted after delete topic is complete");

            // Ensure that the topic is removed from all cleaner offsets
            TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
                    topicPartitions.stream().allMatch(tp -> {
                        List<File> liveLogDirs = CollectionConverters.asJava(broker.logManager().liveLogDirs());
                        return liveLogDirs.stream().allMatch(logDir -> {
                            OffsetCheckpointFile checkpointFile;
                            try {
                                checkpointFile = new OffsetCheckpointFile(new File(logDir, "cleaner-offset-checkpoint"), null);
                            } catch (IOException e) {
                                throw new RuntimeException(e);
                            }
                            return !checkpointFile.read().containsKey(tp);
                        });
                    })),
                "Cleaner offset for deleted partition should have been removed");

            // Ensure that the topic directories are soft-deleted
            TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
                    broker.config().logDirs().stream().allMatch(logDir ->
                        topicPartitions.stream().noneMatch(tp ->
                            new File(logDir, tp.topic() + "-" + tp.partition()).exists()))),
                "Failed to soft-delete the data to a delete directory");

            // Ensure that the topic directories are hard-deleted
            TestUtils.waitForCondition(() -> brokers.stream().allMatch(broker ->
                broker.config().logDirs().stream().allMatch(logDir ->
                    topicPartitions.stream().allMatch(tp ->
                        Arrays.stream(Objects.requireNonNull(new File(logDir).list())).noneMatch(partitionDirectoryName ->
                            partitionDirectoryName.startsWith(tp.topic() + "-" + tp.partition()) &&
                                partitionDirectoryName.endsWith(UnifiedLog.DELETE_DIR_SUFFIX)))
                )
            ), "Failed to hard-delete the delete directory");
        }
    }

    default List<Authorizer> authorizers() {
        List<Authorizer> authorizers = new ArrayList<>();
        authorizers.addAll(brokers().values().stream()
                .filter(server -> server.authorizerPlugin().isDefined())
                .map(server -> server.authorizerPlugin().get().get()).toList());
        authorizers.addAll(controllers().values().stream()
                .filter(server -> server.authorizerPlugin().isDefined())
                .map(server -> server.authorizerPlugin().get().get()).toList());
        return authorizers;
    }


Review comment (suggestion):
Perhaps it should be renamed to `waitTopicCreation` to correspond with `waitTopicDeletion`.

Code after:
        createTopic(topicName, partitions, replicas, Map.of());
    }

    default void createTopic(String topicName, int partitions, short replicas, Map<String, String> props) throws InterruptedException {
        try (Admin admin = admin()) {
            admin.createTopics(List.of(new NewTopic(topicName, partitions, replicas).configs(props)));
            waitForTopic(topicName, partitions);
        }
    }

    /**
     * Deletes a topic and waits for the deletion to complete.
     *
     * @param topicName The name of the topic to delete
     * @throws InterruptedException If the operation is interrupted
     */
    default void deleteTopic(String topicName) throws InterruptedException, ExecutionException {
        try (Admin admin = admin()) {
            admin.deleteTopics(List.of(topicName));
            waitTopicDeletion(topicName);
        }
    }

    void waitForReadyBrokers() throws InterruptedException;

    default void waitForTopic(String topic, int partitions) throws InterruptedException {
        if (partitions <= 0) {
            throw new IllegalArgumentException("Partition count must be > 0, but was " + partitions);
        }

        // wait for metadata
        Collection<KafkaBroker> brokers = aliveBrokers().values();
        TestUtils.waitForCondition(
            () -> brokers.stream().allMatch(broker -> broker.metadataCache().numPartitions(topic).filter(p -> p == partitions).isPresent()),
                60000L, topic + " metadata not propagated after 60000 ms");

        for (ControllerServer controller : controllers().values()) {
            long controllerOffset = controller.raftManager().replicatedLog().endOffset().offset() - 1;
            TestUtils.waitForCondition(
                () -> brokers.stream().allMatch(broker -> ((BrokerServer) broker).sharedServer().loader().lastAppliedOffset() >= controllerOffset),
                60000L, "Timeout waiting for controller metadata propagating to brokers");
        }
    }

    default List<Authorizer> authorizers() {
        List<Authorizer> authorizers = new ArrayList<>();
        authorizers.addAll(brokers().values().stream()
                .filter(server -> server.authorizerPlugin().isDefined())
                .map(server -> server.authorizerPlugin().get().get()).toList());
        authorizers.addAll(controllers().values().stream()
                .filter(server -> server.authorizerPlugin().isDefined())