Spring Integration Kafka消息ID头设置与JdbcMessageStore存储报错处理

问题重现 #

在Spring Integration项目中集成JdbcMessageStore存储Kafka消息时，遇到两个核心问题：

1. 消息无ID头触发报错：

Caused by: java.lang.IllegalArgumentException: Cannot store messages without an ID header
    at org.springframework.util.Assert.notNull(Assert.java:201) ~[spring-core-5.2.15.RELEASE.jar:5.2.15.RELEASE]
    at org.springframework.integration.jdbc.store.JdbcMessageStore.addMessage(JdbcMessageStore.java:314) ~[spring-integration-jdbc-5.3.8.RELEASE.jar:5.3.8.RELEASE]

2. 尝试通过工具设置ID头后，接收端收到的ID为字节数组，触发类型不匹配：

IllegalArgumentException Incorrect type specified for header 'id'. Expected [UUID] but actual type is [B]

原因在于：Kafka生产者无法手动设置id头（属于框架保留头），第三方工具发送的ID头会被序列化为字节数组，无法被JdbcMessageStore识别为UUID类型。

解决方案 #

方案1：接收消息后自动生成UUID类型ID头 #

在Kafka消息进入JdbcMessageStore之前，通过Transformer组件为无ID的消息生成UUID并设置到MessageHeaders.ID头中，确保符合JdbcMessageStore的要求。

代码示例：

@Bean
public GenericTransformer<Message<?>, Message<?>> messageIdEnricher() {
    return incomingMsg -> {
        // 检查消息是否已有合法的UUID类型ID
        if (incomingMsg.getHeaders().getId() == null) {
            return MessageBuilder.fromMessage(incomingMsg)
                    .setHeader(MessageHeaders.ID, UUID.randomUUID())
                    .build();
        }
        return incomingMsg;
    };
}

在IntegrationFlow中配置该Transformer，放在Kafka消息接收组件之后：

@Bean
public IntegrationFlow kafkaMessageFlow(KafkaMessageDrivenChannelAdapter<String, String> kafkaAdapter) {
    return IntegrationFlow.from(kafkaAdapter)
            .transform(messageIdEnricher()) // 先补全ID头
            .handle(/* 后续处理，比如存入JdbcMessageStore的组件 */)
            .get();
}

方案2：转换字节数组类型的ID头为UUID #

如果需要兼容第三方工具发送的ID头，可以自定义KafkaHeaderMapper，将收到的字节数组类型id头转换为UUID。

代码示例：

@Bean
public KafkaHeaderMapper customKafkaHeaderMapper() {
    DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
    // 自定义id头的转换逻辑
    mapper.addHeaderMapper("id", (headerValue, headers) -> {
        if (headerValue instanceof byte[]) {
            try {
                String uuidStr = new String((byte[]) headerValue, StandardCharsets.UTF_8);
                return UUID.fromString(uuidStr);
            } catch (IllegalArgumentException e) {
                // 转换失败时生成新的UUID
                return UUID.randomUUID();
            }
        }
        return headerValue;
    });
    return mapper;
}

将该HeaderMapper配置到Kafka消息驱动适配器：

@Bean
public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter(
        ConsumerFactory<String, String> consumerFactory) {
    KafkaMessageDrivenChannelAdapter<String, String> adapter = 
            new KafkaMessageDrivenChannelAdapter<>(consumerFactory, ListenerMode.record, "your-topic");
    adapter.setHeaderMapper(customKafkaHeaderMapper()); // 设置自定义HeaderMapper
    return adapter;
}

推荐方案 #

优先选择方案1，因为它从根源上保证所有进入JdbcMessageStore的消息都具备合法的UUID类型ID头，无需依赖外部生产者的设置，兼容性和稳定性更强。

内容的提问来源于stack exchange，提问作者Oleksandr H