twitter / finatra

Fast, testable, Scala services built on TwitterServer and Finagle
https://twitter.github.io/finatra/
Apache License 2.0
2.28k stars 409 forks source link

HttpClientModule invoke ChannelClosedException #509

Closed junhoi-pk closed 5 years ago

junhoi-pk commented 5 years ago

HttpClientModule rarely throws a ChannelClosedException. The finatra version I use is 18.1.0 (But version 18.12.0 also had issues.)

It didn't happen when there was a lot of traffic, and the remote server didn't have a problem.

Below is the stack trace

com.twitter.finagle.ChannelClosedException: ChannelException at remote address: xxxxx/x.x.x.x:80 from service: xxxx. Remote Info: Upstream Address: Not Available, Upstream id: Not Available, Downstream Address: xxxxx/x.x.x.x:80, Downstream label: xxxx, Trace Id: 9060cbe8409235ee.58bcbe54c829e4d1<:9060cbe8409235ee
    at com.twitter.finagle.netty4.transport.ChannelTransport$$anon$1.channelInactive(ChannelTransport.scala:188)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at io.netty.handler.codec.MessageAggregator.channelInactive(MessageAggregator.java:417)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at io.netty.handler.codec.http.HttpContentDecoder.channelInactive(HttpContentDecoder.java:205)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelInactive(CombinedChannelDuplexHandler.java:420)
    at io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:377)
    at io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:342)
    at io.netty.handler.codec.http.HttpClientCodec$Decoder.channelInactive(HttpClientCodec.java:281)
    at io.netty.channel.CombinedChannelDuplexHandler.channelInactive(CombinedChannelDuplexHandler.java:223)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at com.twitter.finagle.netty4.channel.ChannelRequestStatsHandler.channelInactive(ChannelRequestStatsHandler.scala:41)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75)
    at com.twitter.finagle.netty4.channel.ChannelStatsHandler.channelInactive(ChannelStatsHandler.scala:136)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:224)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1354)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:245)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:231)
    at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:917)
    at io.netty.channel.AbstractChannel$AbstractUnsafe$8.run(AbstractChannel.java:822)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:403)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:463)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at com.twitter.finagle.util.BlockingTimeTrackingThreadFactory$$anon$1.run(BlockingTimeTrackingThreadFactory.scala:23)
    at java.lang.Thread.run(Thread.java:748)
junhoi-pk commented 5 years ago
lazy val xxxClient = XXXHttpClientBuilder.build(
  conf.getString("services.xxx.label"),
  conf.getString("services.xxx.host"),
  conf.getInt("services.xxx.port"),
  conf.getBoolean("services.xxx.ssl")
)

-----------------------------------------------

object XXXHttpClientBuilder {
  def build(label: String, host: String, port: Int = 80, ssl: Boolean = false, requestTimeout: Duration = 3.seconds) = {
    val httpClientModule = new HttpClientModule {
      override def dest = s"$host:$port"
      override def defaultHeaders: Map[String, String] = Map("Host" -> host)
      def buildHttpClient(mapper: FinatraObjectMapper, httpService: Service[Request, Response]): XXXHttpClient = {
        new XXXHttpClient(
          hostname = hostname,
          httpService = httpService,
          retryPolicy = retryPolicy,
          defaultHeaders = defaultHeaders,
          mapper = mapper
        )
      }
    }
    val httpClient = ssl match {
      case false =>
        Http.client
          .withLabel(label)
          .withTracer(XXXServerMain.injector.instance[HttpZipkinTracer])
          .withSessionQualifier.noFailFast
          .withRequestTimeout(requestTimeout)
          .newClient(s"$host:$port")
          .toService
      case true =>
        Http.client
          .withLabel(label)
          .withTracer(XXXServerMain.injector.instance[HttpZipkinTracer])
          .withSessionQualifier.noFailFast
          .withRequestTimeout(requestTimeout)
          .withTls(host)
          .newClient(s"$host:$port")
          .toService
    }
    val masked = new MaskCancelFilter[Request, Response]
    val maskedClient = masked.andThen(httpClient)
    httpClientModule.buildHttpClient(
      XXXServerMain.injector.instance[FinatraObjectMapper],
      masked.andThen(maskedClient)
    )
  }
}

-----------------------------------------------

class XXXHttpClient(
    hostname: String = "",
    httpService: Service[Request, Response],
    retryPolicy: Option[RetryPolicy[Try[Response]]] = None,
    defaultHeaders: Map[String, String] = Map(),
    mapper: FinatraObjectMapper) extends HttpClient(hostname, httpService, retryPolicy, defaultHeaders, mapper) {

  override def execute(request: Request): Future[Response] = {
    val res: Future[Response] = super.execute(request)
    res.map {
      httpResponse =>
        if(httpResponse.status == Status.InternalServerError)
          MDC.put("errorResponse", s"uri:${request.uri}, code:${httpResponse.statusCode}, body:${httpResponse.contentString} ")
    }
    res
  }
}
cacoco commented 5 years ago

@jungrammer thanks for the issue. Technically, this is something with Finagle you're experiencing. Channel closed generally indicates the connection has been closed. This could happen for any myriad of reasons not all of which are bad. Were you experiencing any network issues at the time?

junhoi-pk commented 5 years ago

@cacoco Thank you for the answer. It's not an error that's often happening, but because it's happening occasionally, I can only see it with error reporting.

cacoco commented 5 years ago

@jungrammer you may be experiencing something similar to https://github.com/twitter/finagle/issues/517 which has come up again recently internally as well as we suspect that we may not be properly handling the Connection: Close header. Any more information you can provide around the system when you observe this behavior would be helpful. Thanks!

roanta commented 5 years ago

@jungrammer as @cacoco mentioned, ChannelClosedException is generic and could indicate that the server has hung up / close the session. Do you have any visibility on the remote server to try to correlate these errors with any such event on the server?

junhoi-pk commented 5 years ago

@roanta Hi roanta.

I've tested when the remote server died before. I've tested when the remote server died before. In other words, we tested the session disconnection on the remote server.

At that time, a ChannelClosedException occurred, but it showed a different stack trace.

Guessing it seems to be caused by the http client closing the connection and attempting to close the session but it is already closed.

In the code below, the above error occurred about 2%.

lazy val xxxClient = XXXHttpClientBuilder.build(
  conf.getString("services.xxx.label"),
  conf.getString("services.xxx.host"),
  conf.getInt("services.xxx.port"),
  conf.getBoolean("services.xxx.ssl")
)

def xxxx(...) = {
    val cli = xxxClient;

    cli.executeJson[XXX](Request("xxx")) 
  }
bryce-anderson commented 5 years ago

@jungrammer, we're unable to reproduce this locally so I'm going to close this for now. If you could write a failing unit test please do post it and reopen the ticket.