Advanced Patterns · Advanced

ConsumerSeekAware — Offset Control

ConsumerSeekAware gives fine-grained control over consumer position. Useful for replaying specific time ranges, starting from a known position, or implementing custom seek logic.

SeekableOrderConsumer.javajava

@Service
@Slf4j
public class SeekableOrderConsumer implements ConsumerSeekAware {

    // Called after partition assignment — seek to a specific position
    @Override
    public void onPartitionsAssigned(
        Map<TopicPartition, Long> assignments,
        ConsumerSeekCallback callback
    ) {
        // Option 1: Seek to timestamp (reprocess last 1 hour)
        long oneHourAgo = System.currentTimeMillis() - 3_600_000;
        assignments.keySet().forEach(tp ->
            callback.seekToTimestamp(tp.topic(), tp.partition(), oneHourAgo));

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

        // Option 3: Seek to 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) {}
}