CODEWITHSUNDEEP
javaspring-bootapache-kafkaspring-kafka
vicky
vicky

Reputation: 940

how to pause and resume @KafkaListener using spring-kafka

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 {

Upvotes: 13

Views: 16494

Answers (1)

bistros
bistros

Reputation: 1179

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

Upvotes: 23

Related Questions