how to get KafkaListener Consumer

Question

I use spring-kafka for springboot 2.0.4.RELEASE.

And use KafkaListener for get message

Now I want to reset the offset for my group

But i do not how to get the consumer for the group

    @KafkaListener(id="test",topics={"test"},groupId="group",containerFactory="batchContainerFactory")
    public String listenTopic33(List<ConsumerRecord<Integer, String>> record, Acknowledgment ack){
// do something

    }


    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    public void test() {

        MessageListenerContainer test3 = kafkaListenerEndpointRegistry.getListenerContainer("test3");
}

Answer 1

If you want to seek the consumer in the listener itself, simply add a Consumer<?, ?> consumer parameter to the listener method.

Bear in mind that the container may have fetched more messages so you will get them before the seek takes affect. You could set max.poll.records=1 to avoid that.

You can also add a custom RemainingRecordsErrorHandler to the container, throw an exception in the listener, and the error handler will get the remaining records instead of the listener.

Also see Seeking to a Specific Offset.

In order to seek, your listener must implement ConsumerSeekAware, which has the following methods:

void registerSeekCallback(ConsumerSeekCallback callback);

void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

The first method is called when the container is started. You should use this callback when seeking at some arbitrary time after initialization. You should save a reference to the callback. If you use the same listener in multiple containers (or in a ConcurrentMessageListenerContainer), you should store the callback in a ThreadLocal or some other structure keyed by the listener Thread.

When using group management, the second method is called when assignments change. You can use this method, for example, for setting initial offsets for the partitions, by calling the callback. You must use the callback argument, not the one passed into registerSeekCallback. This method is never called if you explicitly assign partitions yourself. Use the TopicPartitionInitialOffset in that case.

The callback has the following methods:

void seek(String topic, int partition, long offset);

void seekToBeginning(String topic, int partition);

void seekToEnd(String topic, int partition);

You can also perform seek operations from onIdleContainer() when an idle container is detected. See Detecting Idle and Non-Responsive Consumers for how to enable idle container detection.

To arbitrarily seek at runtime, use the callback reference from the registerSeekCallback for the appropriate thread.

Here's an example; we keep track of the callbacks for each topic/partition...

@SpringBootApplication
public class So56584233Application {

    public static void main(String[] args) {
        SpringApplication.run(So56584233Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(Listener listener, KafkaTemplate<String, String> template) {
        return args -> {
            IntStream.range(0, 10).forEach(i -> template.send(new ProducerRecord<>("so56584233", i % 3, "foo", "bar")));
            while (true) {
                System.in.read();
                listener.seekToStart();
            }
        };
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("so56584233", 3, (short) 1);
    }

}

@Component
class Listener implements ConsumerSeekAware {


    private static final Logger logger = LoggerFactory.getLogger(Listener.class);


    private final Map<TopicPartition, ConsumerSeekCallback> callbacks = new ConcurrentHashMap<>();

    private static final ThreadLocal<ConsumerSeekCallback> callbackForThread = new ThreadLocal<>();

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        callbackForThread.set(callback);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        assignments.keySet().forEach(tp -> this.callbacks.put(tp, callbackForThread.get()));
    }

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

    @KafkaListener(id = "so56584233", topics = "so56584233", concurrency = "3")
    public void listen(ConsumerRecord<String, String> in) {
        logger.info(in.toString());
    }

    public void seekToStart() {
        this.callbacks.forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
    }

}