java - 如何停止@Bean注释的Kstream kafka消费者并继续运行spring-boot应用程序

Question

我在同一个类中有多个@Bean注释的Kstream、kafka消费者，一旦所有消费者消费了特定的消息(K，V)对，我需要停止所有消费者并执行一段单独的代码。//评论部分是我需要的。以下是其中一个消费者的伪代码:

 @Bean
 public KStream<KEY, VALUE> ssStreams(StreamsBuilder builder) {



        KStream<KEY, VALUE> memberKStream = builder.stream(topicClass.getTopicName());

            KStream stream = memberKStream.map((key, value) -> {
                log.info("Consuming key  :::" + key.toString());


                if(value.getItem ='X' )
                {
                    // need to stop this consumer as soon as this condition is true
                   // jump the application to different class of project 

                }            

           }

}

我尝试过的事情:
1. 创建异常并通过 DeserializationExceptionHandler 进行处理。

Answer 1

您应该停止 Kafka 监听器容器