KafkaMessageSource生成无Id消息，能否配置使其生成带Id的消息？

问题本质 #

JdbcChannelMessageStore依赖消息的id头作为存储主键，而KafkaMessageSource默认生成的Message实例不会自动填充这个头信息，导致无法持久化。该行为完全可以通过配置解决。

1. 直接给KafkaMessageSource添加Id生成后置处理器 #

这是最直接的方案，通过IdGeneratingMessagePostProcessor自动为每个从Kafka拉取的消息生成UUID作为Id：

@Bean
public KafkaMessageSource<String, String> kafkaMessageSource(ConsumerFactory<String, String> consumerFactory) {
    KafkaMessageSource<String, String> source = new KafkaMessageSource<>(
        consumerFactory, 
        new ConsumerProperties("your-kafka-topic")
    );
    // 添加Id生成后置处理器
    source.setMessagePostProcessors(new IdGeneratingMessagePostProcessor());
    return source;
}

如果使用@InboundChannelAdapter注解定义源，同样可以在内部配置：

@InboundChannelAdapter(channel = "kafkaInputChannel", poller = @Poller(fixedDelay = "1000"))
public MessageSource<String> kafkaInboundSource(ConsumerFactory<String, String> consumerFactory) {
    KafkaMessageSource<String, String> source = new KafkaMessageSource<>(
        consumerFactory, 
        new ConsumerProperties("your-kafka-topic")
    );
    source.setMessagePostProcessors(new IdGeneratingMessagePostProcessor());
    return source;
}

2. 修复你的MessageTransformer问题 #

你之前尝试用Transformer但未生成Id，是因为默认Transformer不会自动添加Id头。需要在Transformer中手动构建消息时显式设置Id：

@Transformer(inputChannel = "kafkaInputChannel", outputChannel = "storeChannel")
public Message<?> enrichMessageWithId(Message<String> originalMessage) {
    return MessageBuilder.fromMessage(originalMessage)
            // 仅当Id不存在时添加，避免覆盖已有Id
            .setHeaderIfAbsent(IntegrationMessageHeaderAccessor.ID, UUID.randomUUID().toString())
            .build();
}

或者，也可以给Transformer的输出通道配置全局的Id生成处理器，无需修改Transformer逻辑：

@Bean
public MessageChannel storeChannel() {
    return MessageChannels.direct()
            .interceptor(new IdGeneratingMessagePostProcessor())
            .get();
}

3. 全局配置消息Id生成器 #

如果希望所有Spring Integration生成的消息都自动带Id，可以配置全局的MessageIdGenerator：

@Configuration
public class IntegrationGlobalConfig {
    @Bean
    public MessageIdGenerator messageIdGenerator() {
        // 使用UUID作为默认Id生成规则
        return UUID.randomUUID()::toString;
    }
}

注意：KafkaMessageSource默认不会触发全局生成器，因此建议结合方案1使用，确保拉取的Kafka消息能被正确处理。

内容的提问来源于stack exchange，提问作者Milen Manov