How to read from the end of a topic, regardless of the group's committed offset

Question

I am using following packages to consume kafka messages

compile 'org.springframework.boot:spring-boot-starter-webflux'
compile("org.springframework.boot:spring-boot-starter-web")
// tag::actuator[]
compile("org.springframework.boot:spring-boot-starter-actuator")
compile('org.springframework.kafka:spring-kafka:2.1.7.RELEASE')
compile 'io.projectreactor.kafka:reactor-kafka:1.0.0.RELEASE'

I want to consume messages from the end of a topic, regardless of the group's committed offset

When I search I found we can do using following code

consumer = new KafkaConsumer<>(properties);
consumer.seekToEnd(Collections.emptySet());

But I am unable to find how to use above code in spring boot webflux

@Component
public class EventConsumer
{
    private final EmitterProcessor<ServerSentEvent<String>> emitter = EmitterProcessor.create();

    public Flux<ServerSentEvent<String>> get()
    {
        return emitter;
    }

    @KafkaListener(topics = "${kafka.zone.status.topic.name}")
    public void receive(String data)
    {
        //System.out.println(data);
        emitter.onNext(ServerSentEvent.builder(data).id(UUID.randomUUID().toString()).build());
    }
}

Answer 1

See the documentation.

Implement ConsumerSeekAware and perform the seeks in the onPartitionsAssigned method.

@Component
public class EventConsumer implements ConsumerSeekAware {

    private final EmitterProcessor<ServerSentEvent<String>> emitter = EmitterProcessor.create();

    public Flux<ServerSentEvent<String>> get() {
        return emitter;
    }

    @KafkaListener(topics = "${kafka.zone.status.topic.name}")
    public void receive(String data) {
        // System.out.println(data);
        emitter.onNext(ServerSentEvent.builder(data).id(UUID.randomUUID().toString()).build());
    }

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {

    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        assignments.keySet().forEach(tp -> callback.seekToEnd(tp.topic(), tp.partition()));
    }

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

    }

}