Using vertx-spring-boot-starter-kafka, when using seekToBeginning when partition is assigned, the consumer sometimes reads the same message multiple times.

In my situation I have single topic with 2 messages (message1, message2). Reading it when openshift pod starts up, using following code:

consumer.partitionsAssignedHandler(partitions -> {
    Flux.fromIterable(partitions)
        .flatMap(consumer::seekToBeginning)
        // Time out in case this takes too long
        .take(Duration.ofSeconds(2))
        .subscribe();
    });

consumer.subscribe(TOPIC_NAME).block();

disposableConsumer = consumer.flux()
    .subscribe(message -> {
        System.out.println("Received message: " + message);
        messages.add(message);
    });

This code should reset partition to the beginning and then read all messages in the topic.
Sometimes it reads message 2 and then message 1 and message 2 (getting 3 messages in total, although there are only 2 in total).

Attaching full log from the pod. But it seems the kafka consumer works like:

1. Connect to topic
2. read some messages
3. seekToBeginning
4. read all messages again

It should read the messages only once, and that's after it seekToBeginning is finished.