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

Code before:
     */
    default void deleteTopic(String topicName) throws InterruptedException {
        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);
        }

        else if(partitions == 0){
            waitTopicDeletion(topic);
        }

        // wait for metadata
        Collection<KafkaBroker> brokers = aliveBrokers().values();
        TestUtils.waitForCondition(
            () -> brokers.stream().allMatch(broker -> partitions == 0 ?
                broker.metadataCache().numPartitions(topic).isEmpty() :
                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())
                .map(server -> server.authorizerPlugin().get().get()).toList());
        return authorizers;
    }

    default void waitAcls(AclBindingFilter filter, Collection<AccessControlEntry> entries) throws InterruptedException {
        for (Authorizer authorizer : authorizers()) {
            AtomicReference<Set<AccessControlEntry>> actualEntries = new AtomicReference<>(new HashSet<>());
            TestUtils.waitForCondition(() -> {
                Set<AccessControlEntry> accessControlEntrySet = new HashSet<>();

Review comment (suggestion):
line 294: `() -> brokers.stream().allMatch(broker -> partitions == 0 ?`
Here, `partitions == 0` is used to check whether it is waiting for deletion. And the similar "wait for metadata" logic should also be applied to waitTopicDeletion. Please update the code in both methods accordingly.

Code after:
        }
    }

    /**
     * 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())
                .map(server -> server.authorizerPlugin().get().get()).toList());
        return authorizers;
    }

    default void waitAcls(AclBindingFilter filter, Collection<AccessControlEntry> entries) throws InterruptedException {
        for (Authorizer authorizer : authorizers()) {
            AtomicReference<Set<AccessControlEntry>> actualEntries = new AtomicReference<>(new HashSet<>());