使用Kafka Binder在Spring云中打印JsonObject

Question

我是春云和Kafka流的新手。我正在尝试使用 kafka 活页夹设置Spring云应用程序。我尝试在本地测试 kafka 流处理器，但我无法打印任何日志。

我的kafka消息将包含JSONObject。kafkaStreamListener类是：

@Configuration
public class KafkaStreamListener {

    private static Logger logger = LogManager.getLogger(KafkaStreamListener.class);
    
    //bean for processing autonomous messages 
     @Bean
      public Function<KStream<String, JSONObject>, KStream<String, JSONObject>> autonomousProcessor() {
         System.out.println("start of stream processor%%%%%%%%%%%%%%%%%%%%%**************************");
         logger.info("inside processor");
         return kstream -> kstream.filter((key,value) -> {
         System.out.println(value.toString()); 
         return true;});
             }

Application.properties：

#Processor group with inputs and outputs
spring.cloud.stream.function.definition = autonomousProcessor
spring.cloud.stream.bindings.autonomousProcessor-in-0.destination = INPUT_TOPIC
spring.cloud.stream.bindings.autonomousProcessor-out-0.destination = OUTPUT_TOPIC
spring.cloud.stream.kafka.streams.binder.functions.autonomousProcessor.application-id= autonomousProcessorGroup

问题:在调试模式下，断点直接到达过滤器步骤，然后不执行任何操作。它跳过了记录器和SOP。不知道可能是什么问题。春云版:霍克斯顿。SR11

Answer 1

我认为你所看到的是正确的行为。绑定器将在引导时调用您的函数一次，然后将调用初始SOP和记录器（再次调用一次）。如果在启动应用程序时在第一个SOP或记录器上设置断点，您将看到它们被调用。然后，当Kafka主题接收到数据时，将调用提供的lambda（带过滤器）。过滤器中的内部SOP应记录＜code＞值。toString（）对每个记录调用文件管理器。