public void seekToBeginning()

in pulsar-client-kafka-compat/pulsar-client-kafka_0_9/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java [436:456]

• 18 lines of code
• 4 McCabe index (conditional complexity)

    public void seekToBeginning(TopicPartition... partitions) {
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        if (partitions.length == 0) {
            partitions = consumers.keySet().toArray(new TopicPartition[0]);
        }
        lastCommittedOffset.clear();
        lastReceivedOffset.clear();

        for (TopicPartition tp : partitions) {
            org.apache.pulsar.client.api.Consumer<byte[]> c = consumers.get(tp);
            if (c == null) {
                futures.add(FutureUtil.failedFuture(
                        new IllegalArgumentException("Cannot seek on a partition where we are not subscribed")));
            } else {
                futures.add(c.seekAsync(MessageId.earliest));
            }
        }

        FutureUtil.waitForAll(futures).join();
    }