How to pause/resume on Listener ?

[V-Rajasekar]

Can you please share a code snippet how to invoke the pause( )/resume( ) on the listener instead on a consumer as advised in spring-kafka reference doc as it is not always thread-safe.

I tried injecting or Autowiring the listener in the kafka-consumer but it lead to below cyclic error

The dependencies of some of the beans in the application context form a cycle: because
┌─────┐
| myListener defined in class path resource [no/demo/services/configuration/KafkaConfiguration.class]
↑ ↓
| myMessageConsumer (field org.springframework.kafka.listener.KafkaMessageListenerContainer no.demo.services.MyMessageConsumer.myListener)
└─────┘

[garyrussell]

Please don't use GitHub issues to ask questions; use Stack Overflow or Gitter for questions; issues are for bug reports or new feature requests.

See Pausing/Resuming Listener Containers.

[garyrussell]

@SpringBootApplication
public class Kgh650Application implements ApplicationListener<KafkaEvent> {

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

	@Override
	public void onApplicationEvent(KafkaEvent event) {
		System.out.println(event);
	}

	@Bean
	public ApplicationRunner runner(KafkaListenerEndpointRegistry registry,
			KafkaTemplate<String, String> template) {
		return args -> {
			template.send("kgh650", "foo");
			Thread.sleep(10_000);
			System.out.println("pausing");
			registry.getListenerContainer("kgh650").pause();
			Thread.sleep(10_000);
			template.send("kgh650", "bar");
			Thread.sleep(10_000);
			System.out.println("resuming");
			registry.getListenerContainer("kgh650").resume();
			Thread.sleep(10_000);
		};
	}

	@KafkaListener(id = "kgh650", topics = "kgh650")
	public void listen(String in) {
		System.out.println(in);
	}

}

2018-04-16 09:21:32.510  INFO 53965 --- [   kgh650-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [kgh650-0]
foo
pausing
ConsumerPausedEvent [partitions=[kgh650-0]]
resuming
ConsumerResumedEvent [partitions=[kgh650-0]]
bar

[shankarps]

Thanks for this. It was helpful. It would be great if we can add a code example to the docs also (https://docs.spring.io/spring-kafka/reference/html/_reference.html#pause-resume). There is a different pause() method in the KafkaConsumer API and the doc was a bit ambiguous if it was referring to the "Consumer" object or the Container object.