Spring Integration响应丢失跟进：跨Flow通道异常与响应方式排查

以下是补充Artem Bilan要求的详细信息：

第一个Integration Flow #

@Configuration
@ConditionalOnExpression(" condition evaluates to TRUE ")
public class BeanDefA {

  @Bean
  IntegrationFlow rest() {
    return IntegrationFlow.from((Http.inboundGateway("/myEndpoint")
        .requestMapping(r -> r.methods(HttpMethod.valueOf("POST")))
        .headerMapper(restConfig.httpHeaderMapper())
        .requestPayloadType(String.class)
        .replyTimeout(5000)
        .errorChannel(exceptionConfig.adapterErrorChannel())))

        .handle(hmacValidationService, "validateHMACAuthHeaders")

        .enrichHeaders(h -> h.headerFunction("message-correlation-id", m -> UUID.randomUUID().toString()))

        .publishSubscribeChannel(c -> c

            .subscribe(s -> s

                .routeToRecipients(r -> r

                    .recipient("backupChannelA", m -> true)

                    .recipient("backupChannelB", m -> false)

                    .defaultOutputToParentFlow())

                .channel("nullChannel"))

            .subscribe(s -> s

                .handle(messagePayloadService, "mapMessagePayload")

                .enrich(e -> e.requestChannel("claimCheck.input").headerExpression("claimCheck", "payload"))

                .handle("integrationGateway", "process")))
        .get();
  }
}

其中.handle(messagePayloadService, "mapMessagePayload")关联的Bean包含以下方法，期望通过replyChannel向/myEndpoint的POST请求方返回{"Status":"Success","ErrorMessage":""}响应：

public boolean sendReply(Map<String, Object> headers) {
    MessageChannel replyChannel = (MessageChannel) headers.get("replyChannel");

    Message m = MessageBuilder.withPayload("{\"Status\":\"Success\",\"ErrorMessage\":\"\"}")
       .setHeader(HttpHeaders.STATUS_CODE, response.getStatusCode())
       .setHeader("Content-Type", HEADER_FMT).build();

    return replyChannel.send(m);
}   

第二个Integration Flow #

@Configuration
@ConditionalOnExpression(" condition evaluates to TRUE ")
public class BeanDefB {

  @Bean
  public IntegrationFlow dbWrite() {
    return f -> f

        .<MessageItemPayloadInterface>handle((p, h) -> {

            Object o = writer.filterAndEnrich(p, h);

            if (o == null && h.get(KafkaHeaders.ACKNOWLEDGMENT) != null) {
                return null;
            }

            return o;

        }, e -> e.id("dbWriterFilterAndEnrichEndpoint"))

        .<MessageItemPayloadInterface>handle((p, h) -> {

            writer.save(p, h);
            return p;

        }, e -> e.id("dbWriterSaveEndpoint").transactional(false))

        .enrichHeaders(h -> h.headerFunction(IntegrationHeaders.PRODUCER_EVENT_KEY,
            m -> {
                String header = writer.generateEventHeader(m.getPayload(), m.getHeaders());
                return header == null ? "" : header;
            }))

        .<MessageItemPayloadInterface>handle((p, h) -> {

            Object o = writer.notify(p, h);

            if (o == null && h.get(KafkaHeaders.ACKNOWLEDGMENT) != null) {
                LOG("this is an error");
                h.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();
                return null;
            }

            return o;

        }, e -> e.id("dbWriterNotifyEndpoint"))

        .publishSubscribeChannel(a -> a

            .subscribe(b -> b

                .routeToRecipients(r -> r

                    .recipient("eventProducerChannel", true)

                    .defaultOutputToParentFlow())

                .channel("nullChannel"))

            .subscribe(d -> d

                .handle((p, h) -> {

                    if (h.get(KafkaHeaders.ACKNOWLEDGMENT) != null) {

                        LOG.debug("Message processed! Headers= " + h + ". Payload= " + p);
                        h.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();

                    }
                    return p;
                })

                .channel("nullChannel")))

        .channel("nullChannel");
  }

}

其中writer是包含业务逻辑的Bean，方法执行出错时返回null。当前问题：writer.notify(p, h)的输出本应发送至Kafka，却意外返回给了/myEndpoint的POST请求方。

已开启调试，仅在发送期望响应前的日志中能看到{"Status":"Success","ErrorMessage":""}，后续流程中无相关踪迹。

编排类 #

@Configuration
@IntegrationComponentScan
public class IntegrationProvider {

  public final Config config;

  public IntegrationProvider(Config config) {
    this.config = config;
  }

  @Bean
  public IntegrationFlow main() {
    return f -> f.routeToRecipients(r -> r.recipient(config.getEndpointType() + ".input"));
  }

  @MessagingGateway(name = "integrationGateway", errorChannel = "errorChannel")
  public interface IntegrationGateway {

    @Gateway(requestChannel = "main.input")
    void start(Object msg);

    @Gateway(requestChannel = "process.input")
    void process(Message<MessagePayloadInterface> msg);

    @Gateway(requestChannel = "write.input")
    void write(Message<MessageItemPayloadInterface> msg);

    @Gateway(requestChannel = "dbWrite.input")
    void dbWrite(Message<MessageItemPayloadInterface> msg);

    @Gateway(requestChannel = "send.input")
    void send(Message<MessageItemPayloadInterface> msg);
  }
}

咨询问题 #

• 是否Spring Integration 5到6的版本变更导致第二个Flow共享了第一个Flow的replyChannel？
• 依赖replyChannel返回POST响应的方式是否本身存在问题？

问题解答 #

1. 版本变更是否导致replyChannel共享？ #

Spring Integration 5到6的版本变更不会主动导致不同Flow共享replyChannel，但你的代码逻辑存在上下文传递的问题：

• HTTP入站网关会自动在消息头中注入replyChannel，该通道关联到当前请求的响应链路。
• 当你通过integrationGateway.process()转发消息时，原始消息的replyChannel头会被完整携带到dbWrite Flow中。
• 由于dbWrite Flow没有清除或替换这个头，当writer.notify()返回结果后，该结果会沿着replyChannel直接返回给HTTP请求方，而非流向Kafka通道。

2. 依赖replyChannel返回响应的方式是否存在问题？ #

这种方式存在明显的设计缺陷：

• HTTP入站网关本身是同步请求-响应模式，网关会自动监听replyChannel等待响应，无需手动调用replyChannel.send()。手动发送的操作会和网关的自动响应逻辑冲突，导致响应被覆盖或异常传递。
• 手动发送响应后，后续异步流程如果仍持有该replyChannel头，其输出会错误流入原始响应链路，出现你遇到的问题。
• 正确做法是：让HTTP入站网关的Flow通过正常流程返回响应（比如在mapMessagePayload中直接返回成功响应体），由网关自动完成响应发送。

修复建议 #

1. 移除手动发送replyChannel的逻辑：修改mapMessagePayload方法，直接返回{"Status":"Success","ErrorMessage":""}作为该handle的输出，HTTP网关会自动将其作为响应返回给请求方。
2. 隔离异步流程的上下文：在integrationGateway.process()发送消息前，或者在dbWrite Flow的起始位置，添加enrichHeaders(h -> h.remove("replyChannel"))，清除原始响应通道头，避免后续流程的结果流入HTTP响应链路。
3. 明确异步处理边界：如果dbWrite是异步业务逻辑，建议使用QueueChannel或异步通道配置，确保同步请求上下文不会渗透到异步流程中。

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