how to pause and resume @KafkaListener using spring-kafka

4
votes

I have implemented the Kafka consumer, now I have a scenario.

  1. Read data from the Kafka stream 2.2.5.Release via Srpingboot
  2. load in the database table1
  3. copy the data from table1 to table2
  4. clear the table1

To do the above things, I need to pause/resume the Kafka consumer using a scheduling job(already written) using quartz, which copies data from table 1 to table 2. But during this activity, I want my Kafka listener to pause, and once the copy is done, it should resume.

My implementation:

@KafkaListener(topicPartitions =
    { @TopicPartition(topic = "data_pipe", partitions = { "0" })})
public void listen(ConsumerRecord<String, String> cr) throws Exception {
1
javaspring-bootapache-kafkaspring-kafka

1 Answers

6
votes

if you use 'kafkaListener annotation' auto created KafkaListenerEndpointRegistry bean, so, You can use it like this code:

@Component
public class KafkaManager {

    private final KafkaListenerEndpointRegistry registry;

    public KafkaManager(KafkaListenerEndpointRegistry registry) {
        this.registry = registry;
    }
    public void pause() {
        registry.getListenerContainers().forEach(MessageListenerContainer::pause);
    }

    public void resume() {
        registry.getListenerContainers().forEach(MessageListenerContainer::resume);
    }
}

document : https://docs.spring.io/spring-kafka/reference/html/#pause-resume