Hello guys, I'm using KafkaSource to define the Kafka consumer:

KafkaSource
        .<...>builder()
        .setProperty("<http://partition.discovery.interval.ms|partition.discovery.interval.ms>", "10000")
        .setProperty("enable.auto.commit", "true")
        .setProperty("<http://auto.commit.interval.ms|auto.commit.interval.ms>", "50")
        .setProperty("auto.offset.reset", "latest")
        .setProperty("auto.offset.reset.strategy", "latest")
        .setProperty("commit.offsets.on.checkpoint", "false")
        .setProperty("client.id", "clientFlink6661")
        .setPartitions(
                new HashSet<>(
                        Arrays.asList(
                                new TopicPartition("myTopic", 0),
                                new TopicPartition("myTopic", 1)
                        )
                )
        )
        .setBootstrapServers(kafkaAddress)
        .setStartingOffsets(OffsetsInitializer.latest())
        .setGroupId("FlinkDestroyer6661")
        .setValueOnlyDeserializer(new MyClassDeserializationSchema())
        .build();

The issue is when I used the

DeliveryGuarantee.EXACTLY_ONCE

in my KafkaSink, everything worked perfectly but when I switched to

DeliveryGuarantee.AT_LEAST_ONCE)

now every time I start the Flink app, it starts from the first message in the topic, like the OffsetsInitializer.latest() doesn't matter at all. I've tried everything like you can see in the provided code. Any ideas what could cause the issue. Also, even though the flink job consumes messages from the Kafka topic, it shows 0 active consumers and the State is marked as EMPTY.

IIRC

OffsetsInitializer

is only for getting the initial offsets (as in, you've never read from the Kafka partitions before). Since you have done that but are only switching the delivery guarantee, you're basically resuming with your read. To be honest, I don't know if you can change your delivery guarantees on the same consumer; I think that might be a breaking change

Thank you for the answer. But the issue where the active consumers are set as 0 still persists. I've tried literally everything

Have you tried as a new consumer?