如何使用自定义元素测试扩展的 kafkaListener？答案

【问题标题】：How to test extended kafkaListener with custom element?如何使用自定义元素测试扩展的 kafkaListener？
【发布时间】：2022-11-16 22:25:50
【问题描述】：

我的开发环境： Java 8 弹簧卡夫卡 - 2.6.1 spring-kafka-test- 2.6.7 弹簧启动 - 2.3.10

用于发送消息的 Rest API

@PostMapping(
        value = {"publish"})
@ExternalApi
public ResponseEntity<Response> endpoint() {

    Response response= new Response();
    User user= new User();
    user.setName("userfirstname");
    LOG.info("in controller ->" + event);
    kafkaTemplate.send("user.topic","sendign String message");
    return new ResponseEntity<>(response,
                                HttpStatus.OK);
}

消费者

@myListener(topics = "user.topic", myattr="user.topic.deadletter")
public void consume(ConsumerRecord<?, User> consumerRecord) {
    LOG.info("consumer topic-> " + consumerRecord.topic());
    LOG.info("consumer value-> " + consumerRecord.value());
}

我有元监听器

@KafkaListener(
        containerFactory = "listenerContainerFactory",
        autoStartup = "false",
        properties = {
                }
)
public @interface myListener {
    @AliasFor(
            annotation = KafkaListener.class,
            attribute = "groupId")
    String groupId() default "";

    String myattr() default "";
}

现在我想向 myattr="user.topic.deadletter" 中提到的新主题发送消息，以防出现问题？

如何将消息发送到 myattr 中提到的以及如何从自定义注释@myListener 中获取值

【问题讨论】：

标签： java spring spring-kafka dead-letter

【解决方案1】：

您无法在运行时访问注释属性；一种解决方案是添加 BeanPostProcessor 以检查注释上的属性并将其作为属性注入到 bean 中。

或者，简单地向 bean 本身添加代码以检查方法注释（例如在构造函数中）并将值存储在字段中。

更高级的解决方案是围绕 bean 创建一个代理，并将属性值作为标头添加到 ConsumerRecord。

【讨论】：