Kafka-Message自定义头中允许的包

Question

在spring-kafka中，如何从包中添加类作为自定义头字段来信任？

消息是这样发送的：

@Autowired
private KafkaTemplate kafkaTemplate;

Message<BouquetMQDTO> m = MessageBuilder
            .withPayload(payload)
            .setHeader(KafkaHeaders.TOPIC, "topic")
            .setHeader("EVENT_TYPE", MessageType.UPSERT)
            .build();
kafkaTemplate.send(m);

接收端如下所示：

@Component
@KafkaListener(topics = "topic")
public class KafkaController {

    @KafkaHandler
    public void listen(
        @Payload Object objectDTO,
        @Header(value = "EVENT_TYPE") MessageType messageType
    ) { 
        System.out.println(messageType);
    }
}

我不断得到的例外是：

@Bean
public KafkaHeaderMapper defaultKafkaHeaderMapper() {
    DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
    mapper.addTrustedPackages("my.package");
    return mapper;
}

props.put(JsonDeserializer.TRUSTED_PACKAGES, "my.other.package,my.package");
return new DefaultKafkaConsumerFactory<>(props);

Answer 1

jsonDeserializer.trusted_packages与头完全无关。它可以处理`ConsumerRecord的键或值。头映射器发生在不同的地方。

不确定是否使用Spring Boot，但是有一个MessagingMessageListenerAdapter，它附带了一个默认的MessagingMessageConverter，因此，默认的DefaultKafkAheaderMapper。要自定义您自己的HeaderMapper，您需要将MessagingMessageConverter、引用HeaderMapper并将该转换器注入到AbstractKafKalistenerContainerFactorybean中。

如果您处理Spring Boot，只需声明MessagingMessageConverter，它将自动配置到框架创建的AbstractKafKalistenerContainerFactory中。

通过这种方式，您可以访问受信任的包。但是，我认为它还不能起作用，因为enum默认情况下对JSON不太友好：https://www.baeldung.com/jackson-serialize-enums