Camel异常处理后抛出Stream Closed IOException问题排查求助

1. Camel异常处理器配置 #

我在Camel中配置了如下异常处理器：

<camel:onException> 
    <camel:exception>com.example.BadDataException</camel:exception>
    <camel:handled>  
        <camel:constant>true</camel:constant>  
    </camel:handled>
    <camel:process ref="badDataErrorProcessor" />  
    <camel:setHeader headerName="Exchange.HTTP_RESPONSE_CODE">  
        <camel:constant>400</camel:constant>  
    </camel:setHeader>  
</camel:onException>

2. 自定义错误处理器实现 #

BadDataErrorProcessor是自定义Bean，主要实现两个功能：

• 记录错误日志
• 生成优雅的错误响应

代码实现如下：

@Component( "badDataErrorProcessor" )  
public class BadDataErrorProcessor implements Processor  
{  

   private static final Logger logger  = LoggerFactory.getLogger( BadDataErrorProcessor.class );  

   @Override  
   public void process( Exchange exchange )  
   {  
      logger.info( "Entering ErrorProcessor : process()" );  

      orderResponse( exchange );  

      logger.debug( "Error Processor setting ID for ..." );  
   }  

   public OrderFulfillmentResponse orderResponse( Exchange exchange )  
   {  
      OrderFulfillmentResponse orderErrorResponse = new OrderFulfillmentResponse();  

      String errorCode = exchange.getIn()  
                                 .getHeader( ERROR_CODE,  
                                             String.class );  

      logger.info( "Inside BadDataErrorProcessor with exchangeID {}:",  
                   exchange.getExchangeId() );  
      Throwable throwable = exchange.getProperty( Exchange.EXCEPTION_CAUGHT,  
                                                  Throwable.class );  

      if( throwable instanceof BadDataException )  
      {  
         BadDataException validationEx = (BadDataException) throwable;  
         logger.debug( "BadDataException, required fields were mising{} :",  
                       validationEx.getMessage() );  
      }  
      OrderGroupResponse orgpRes = new OrderGroupResponse();  
      orderErrorResponse.setStatus( Status.FAILED );  
      orderErrorResponse.setResponseCode( errorCode ); 
      orderErrorResponse.setResponseMessage( throwable.getMessage() );  
      orgpRes.setStatus( Status.FAILED );  
      orgpRes.setResponseCode( errorCode );  
      orgpRes.setResponseMessage( "400 Bad Request" );  
      orderErrorResponse.addOrderGroupResponsesItem( orgpRes );  

      exchange.getIn()  
              .setBody( orderErrorResponse );  
      exchange.getIn()  
              .setHeaders( exchange.getIn()  
                                   .getHeaders() );  

      return orderErrorResponse;  
   }  
}

3. 异常堆栈信息 #

当BadDataErrorProcessor执行完成后，Camel在org.apache.camel.component.servlet.CamelHttpTransportServlet类中记录了如下堆栈：

CamelHttpTransportServlet -  Error processing request
java.io.IOException: Stream closed
    at org.apache.catalina.connector.InputBuffer.throwIfClosed(InputBuffer.java:526) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.connector.InputBuffer.read(InputBuffer.java:337) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.connector.CoyoteInputStream.read(CoyoteInputStream.java:132) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.connector.CoyoteInputStream.read(CoyoteInputStream.java:110) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.camel.util.IOHelper.copy(IOHelper.java:205) ~[camel-core-2.25.4.jar:2.25.4]
    at org.apache.camel.http.common.DefaultHttpBinding.copyStream(DefaultHttpBinding.java:432) ~[camel-http-common-2.22.3.jar:2.22.3]
    at org.apache.camel.http.common.DefaultHttpBinding.doWriteDirectResponse(DefaultHttpBinding.java:496) ~[camel-http-common-2.22.3.jar:2.22.3]
    at org.apache.camel.http.common.DefaultHttpBinding.doWriteResponse(DefaultHttpBinding.java:395) ~[camel-http-common-2.22.3.jar:2.22.3]
    at org.apache.camel.http.common.DefaultHttpBinding.writeResponse(DefaultHttpBinding.java:322) ~[camel-http-common-2.22.3.jar:2.22.3]
    at org.apache.camel.http.common.CamelServlet.doService(CamelServlet.java:223) ~[camel-http-common-2.22.3.jar:2.22.3]
    at org.apache.camel.http.common.CamelServlet.service(CamelServlet.java:78) ~[camel-http-common-2.22.3.jar:2.22.3]
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:764) ~[tomcat-embed-core-9.0.63.jar:4.0.FR]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:227) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53) ~[tomcat-embed-websocket-9.0.63.jar:9.0.63]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100) ~[spring-web-5.3.20.jar:5.3.20]
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117) ~[spring-web-5.3.20.jar:5.3.20]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93) ~[spring-web-5.3.20.jar:5.3.20]
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:117) ~[spring-web-5.3.20.jar:5.3.20]
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:96) ~[spring-boot-actuator-2.6.8.jar:2.6.8]

4. 疑问点 #

• 此处的“Stream”具体指什么？
• 它为何会被关闭？
• Camel为何会在它关闭后尝试读取？

若需要更多排查信息，请告知。

5. 排查更新 #

排查发现核心问题是流刚创建就处于已关闭状态：

• Camel的DefaultHttpBinding组件负责写入直接响应，它生成预期的优雅响应消息后，将其转换为输入流
• 随后组件尝试复制该流并最终关闭它，但在读取流时提示流已关闭
• 该流由Camel自身创建，却被Camel标记为已关闭，疑似Camel内部逻辑问题
• 涉及的输入流是Tomcat的CoyoteInputStream，该类通过内部字段标识流是否关闭，确认流确实在创建后就处于关闭状态

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