ConsumerSeekAware — Offset Control

@Service
@Slf4j
public class SeekableOrderConsumer implements ConsumerSeekAware {

    // Called when partition assignments change: seek to a specific position
    @Override
    public void onPartitionsAssigned(
        Map<TopicPartition, Long> assignments,
        ConsumerSeekCallback callback
    ) {
        // Option 1: seek to a timestamp (reprocess the last hour);
        // this seeks to the first offset greater than or equal to the timestamp
        long oneHourAgo = System.currentTimeMillis() - 3_600_000;
        assignments.keySet().forEach(tp ->
            callback.seekToTimestamp(tp.topic(), tp.partition(), oneHourAgo));

        // Option 2: seek to a specific offset
        // assignments.keySet().forEach(tp -> callback.seek(tp.topic(), tp.partition(), 0));

        // Option 3: seek to the beginning
        // callback.seekToBeginning(assignments.keySet());
    }

    @KafkaListener(topics = "orders.order.created", groupId = "replay-consumer")
    public void process(OrderEvent event) {
        replayService.process(event);
    }

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {}

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments,
                                ConsumerSeekCallback callback) {}
}