获取AggregatingKafkaReplyingTemplate的CorrelationId及解决手动回复验证失败问题

• 基于Spring Boot开发的应用，使用AggregatingKafkaReplyingTemplate实现与下游服务的请求-响应模式
• 本地测试流程：启动Spring Boot生产者应用，确认请求消息发送至生产者主题后，通过Local Kafka UI手动在回复主题发送消息
• 核心需求：在Spring Boot应用内（如日志）记录或获取Kafka模板生成的CorrelationId字符串值（而非字节值），并能用该值在Local Kafka UI中发送回复消息完成关联验证
• 遇到的问题：曾通过调试Kafka模板类获取生成的UUID，在Kafka UI中发送后关联失败，报错：

  No pending reply: perhaps timed out, or using a shared reply topic

• 日志现状：仅记录CorrelationId的字节值哈希和字节本身，无法看到实际UUID字符串

1. 记录CorrelationId字符串到日志 #

AggregatingKafkaReplyingTemplate默认用UUID生成CorrelationId并序列化为字节数组，要获取并记录字符串形式，有两种可行方式：

方式一：自定义CorrelationId生成器 #

实现CorrelationIdStrategy接口，自行生成并记录CorrelationId：

@Component
public class LoggingCorrelationIdStrategy implements CorrelationIdStrategy {

    private static final Logger log = LoggerFactory.getLogger(LoggingCorrelationIdStrategy.class);

    @Override
    public byte[] generateCorrelationId(ProducerRecord<?, ?> producerRecord) {
        String correlationId = UUID.randomUUID().toString();
        log.info("请求生成CorrelationId: {}", correlationId);
        return correlationId.getBytes(StandardCharsets.UTF_8);
    }
}

在配置类中替换默认策略：

@Configuration
public class KafkaConfig {

    @Bean
    public AggregatingKafkaReplyingTemplate<?, ?, ?> aggregatingKafkaReplyingTemplate(
            ProducerFactory<?, ?> producerFactory,
            ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory,
            LoggingCorrelationIdStrategy correlationIdStrategy) {
        AggregatingKafkaReplyingTemplate<?, ?, ?> template = new AggregatingKafkaReplyingTemplate<>(producerFactory, containerFactory);
        template.setCorrelationIdStrategy(correlationIdStrategy);
        return template;
    }
}

方式二：拦截Producer请求反序列化CorrelationId #

通过ProducerInterceptor拦截发送的请求，从消息头取出CorrelationId字节数组转成字符串记录：

@Component
public class CorrelationIdLoggingInterceptor implements ProducerInterceptor<Object, Object> {

    private static final Logger log = LoggerFactory.getLogger(CorrelationIdLoggingInterceptor.class);

    @Override
    public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> record) {
        byte[] correlationIdBytes = record.headers().lastHeader(KafkaHeaders.CORRELATION_ID).value();
        String correlationId = new String(correlationIdBytes, StandardCharsets.UTF_8);
        log.info("发送请求，CorrelationId: {}", correlationId);
        return record;
    }

    // 以下方法默认实现即可
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {}
    @Override
    public void close() {}
    @Override
    public void configure(Map<String, ?> configs) {}
}

在生产者配置中添加拦截器：

spring.kafka.producer.properties.interceptor.classes=com.yourpackage.CorrelationIdLoggingInterceptor

2. 解决手动发送回复的关联失败问题 #

关联失败通常由以下原因导致，对应解决方式如下：

原因1：CorrelationId序列化/反序列化不一致 #

默认情况下，模板将UUID转成UTF-8字节数组作为CorrelationId，手动发送时需确保：

• 在Kafka UI中，将CorrelationId字符串以UTF-8编码填入消息头（部分UI会自动处理编码，需确认逻辑）
• 若自定义了序列化方式，手动发送的编码格式必须与模板一致

原因2：回复主题或分区不匹配 #

• 确认手动发送的回复消息，发送至AggregatingKafkaReplyingTemplate监听的正确回复主题
• 若使用分区策略，需确保回复消息发送到对应分区，或配置模板监听所有分区

原因3：请求超时 #

默认回复超时时间较短，手动操作易触发超时，可延长超时时间：

spring.kafka.reply-template.default-reply-timeout=30000

或在发送请求时指定超时：

RequestReplyFuture<?, ?, ?> future = template.sendAndReceive(record);
future.get(30, TimeUnit.SECONDS);

原因4：共享回复主题的消费者组问题 #

若使用共享回复主题，需确保手动发送的消息能被当前应用的消费者组接收：在Kafka UI中指定消息对应的消费者组，或确认模板的回复容器监听的消费者组可获取该消息

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