Heyp Kafka Retry Mechanism

Scratch.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

class Scratch {

    private final List<String> consumedTopic;
    private final String topic;
    private LocalDateTime shouldWaitUntil;

    public Scratch(String topic) {
        this.topic = topic;
        this.consumedTopic = List.of(this.topic);
    }

    public void execute() {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(Map.of());
        consumer.subscribe(consumedTopic);

        while (true) {
            if (shouldWaitUntil != null && LocalDateTime.now().isBefore(shouldWaitUntil)) {
                consumer.resume(consumedTopic.stream().map(s -> new TopicPartition(s, 0)).collect(Collectors.toList()));
                shouldWaitUntil = null;
            }

            final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(5));
            
            final Iterable<ConsumerRecord<String, String>> records = poll.records(topic);

            for (var record : records) {
                final long minTimestamp = LocalDateTime.now().minusMinutes(15).toEpochSecond(ZoneOffset.UTC);

                if (record.timestamp() < minTimestamp) {
                    // do something with record here
                    consumer.commitSync();
                } else {
                    consumer.pause(consumedTopic.stream().map(s -> new TopicPartition(s, 0)).collect(Collectors.toList()));
                    shouldWaitUntil = LocalDateTime.ofEpochSecond(record.timestamp(), 0, ZoneOffset.UTC).plusMinutes(15);
                    // Peut être pas nécessaire
                    consumer.seek(new TopicPartition(topic, 0), record.offset());
                    return;
                }
            }
        }
    }
}