Spring Integration DSL filter()无限等待及ReplyRequiredException问题

问题现象 #

• 在publishingGateway对应的IntegrationFlow中添加Filter组件后，未配置requiresReply(true)时，消息管道出现无限等待；移除Filter组件后，流程恢复正常。
• 当为Gateway配置spec.requiresReply(true)时，会抛出ReplyRequiredException，异常信息提示“处理器未生成回复”，该异常会被errorHandler123记录。

原因分析 #

Filter组件默认行为是丢弃不符合过滤条件的消息，即不会将这类消息传递到下游流程。而Gateway在未开启requiresReply配置时，会持续等待回复消息返回，若没有任何组件生成回复，就会陷入无限阻塞状态；开启requiresReply(true)后，Gateway会主动检测是否有回复生成，当发现无回复时，会立即抛出ReplyRequiredException，这是符合预期的异常触发逻辑。

解决方案 #

针对该问题，可根据业务需求选择以下处理方案：

1. 为丢弃的消息配置处理通道
   通过discardChannel指定不符合过滤条件消息的处理通道，确保Gateway能收到回复（即使是针对丢弃消息的响应），示例代码：

   .filter(payload -> /* 自定义过滤条件判断 */, 
           spec -> spec.discardChannel("discardMessageChannel"))
   

2. 过滤不通过时直接抛出异常
   若不需要保留不符合条件的消息，可配置Filter在过滤不通过时抛出异常，配合Gateway的异常处理器处理，示例代码：

   .filter(payload -> /* 自定义过滤条件判断 */, 
           spec -> spec.throwExceptionOnRejection(true))
   

3. 设置Gateway超时时间
   为Gateway配置replyTimeout，避免无限等待，超时后会自动终止等待并抛出超时异常，示例代码：

   .gateway(spec -> spec.replyTimeout(5000)) // 设置5秒超时
   

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