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

Code before:
     */
    default int waitUntilLeaderIsElectedOrChangedWithAdmin(Admin admin,
                                                           String topic,
                                                           int partitionNumber,
                                                           long timeoutMs) throws Exception {
        long startTime = System.currentTimeMillis();
        TopicPartition topicPartition = new TopicPartition(topic, partitionNumber);

        while (System.currentTimeMillis() < startTime + timeoutMs) {
            try {
                TopicDescription topicDescription = admin.describeTopics(List.of(topic))
                        .allTopicNames().get().get(topic);

                Optional<Integer> leader = topicDescription.partitions().stream()
                        .filter(partitionInfo -> partitionInfo.partition() == partitionNumber)
                        .findFirst()
                        .map(partitionInfo -> {
                            int leaderId = partitionInfo.leader().id();
                            return leaderId == Node.noNode().id() ? null : leaderId;
                        });

                if (leader.isPresent()) {
                    return leader.get();
                }

            } catch (InterruptedException | ExecutionException ignored) {
                // Continue retrying on any exception (network issues, topic not ready, etc.)
            }

            Thread.sleep(Math.min(100L, timeoutMs));
        }

        throw new AssertionError("Timing out after " + timeoutMs +
                " ms since a leader was not elected for partition " + topicPartition);
    }
}


Review comment (suggestion):
the "retry" mechanism is already in `Admin`, and hence this catch block seems to be redundant.

Code after:

    /**
     * Wait for a leader to be elected or changed using the provided admin client.
     */
    default int waitUntilLeaderIsElectedOrChangedWithAdmin(Admin admin,
                                                           String topic,
                                                           int partitionNumber,
                                                           long timeoutMs) throws Exception {
        long startTime = System.currentTimeMillis();
        TopicPartition topicPartition = new TopicPartition(topic, partitionNumber);

        while (System.currentTimeMillis() < startTime + timeoutMs) {
            try {
                TopicDescription topicDescription = admin.describeTopics(List.of(topic))
                        .allTopicNames().get().get(topic);

                Optional<Integer> leader = topicDescription.partitions().stream()
                        .filter(partitionInfo -> partitionInfo.partition() == partitionNumber)
                        .findFirst()
                        .map(partitionInfo -> {
                            int leaderId = partitionInfo.leader().id();
                            return leaderId == Node.noNode().id() ? null : leaderId;
                        });

                if (leader.isPresent()) {
                    return leader.get();
                }
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof UnknownTopicOrPartitionException ||
                        cause instanceof LeaderNotAvailableException) {
                    continue;
                } else {
                    throw e;
                }
            }

            TimeUnit.MILLISECONDS.sleep(Math.min(100L, timeoutMs));
        }

        throw new AssertionError("Timing out after " + timeoutMs +
                " ms since a leader was not elected for partition " + topicPartition);
    }

    default List<Integer> boundPorts() {
        return brokers().values().stream()
                .map(KafkaBroker::socketServer)
                .map(s -> s.boundPort(clientListener()))
                .toList();

    }