How to catch 'com.linecorp.armeria.common.stream.ClosedStreamException: received a RST_STREAM frame: CANCEL'?

[angela278]

Hi team, we occasionally receive RST_STREAMs from our clients due to timing out before our server is able to send a response back. The messages are expected to be retried so we are not worried about data loss, however we would like to catch the RST_STREAM exceptions and handle them ourselves. I am trying to implement a ServerInteceptor as an exception handler of sorts, however the handleException() function isn't reached when we encounter the exception.

public class GrpcExceptionHandlerInterceptor implements ServerInterceptor {

  @Override
  public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
      ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    ServerCall.Listener<ReqT> delegate = next.startCall(call, headers);
    return new ExceptionHandler<>(delegate, call, headers);
  }

  private static class ExceptionHandler<T, R>
      extends ForwardingServerCallListener.SimpleForwardingServerCallListener<T> {

    private final ServerCall<T, R> delegate;
    private final Metadata headers;

    ExceptionHandler(
        ServerCall.Listener<T> listener, ServerCall<T, R> serverCall, Metadata headers) {
      super(listener);
      this.delegate = serverCall;
      this.headers = headers;
    }

    @Override
    public void onHalfClose() {
      try {
        super.onHalfClose();
      } catch (Exception ex) {
        handleException(ex, delegate, headers);
        throw ex;
      }
    }

    private void handleException(
        Exception exception, ServerCall<T, R> serverCall, Metadata headers) {
      if (exception instanceof StatusRuntimeException || exception instanceof ClosedStreamException) {
          // Do something
      }
    }
  }
}

The full exception stack trace is shown below. I wasn't sure if this was the right place to ask this question or in grpc-java, but the stacktrace shows the exception coming from Armeria so I thought I'd try here first. Am I missing something in order to catch this specific exception? Or is there a better way of handling this? Thank you!

2024-03-04 23:45:38.323 EST WARN  [armeria-common-worker-kqueue-3-2] com.linecorp.armeria.server.logging.LoggingService.(): - [sreqId=c149e21f, chanId=e8a45129, raddr=127.0.0.1:54984, laddr=127.0.0.1:8443][h2c://qtm6j719wy/opentelemetry.proto.collector.trace.v1.TraceService/Export#POST] Response: {startTime=2024-03-05T04:45:38.322Z(1709613938322290), length=47B, duration=814µs(814375ns), totalDuration=2912µs(2912208ns), cause=io.grpc.StatusRuntimeException: CANCELLED, headers=[:status=500, content-type=text/plain; charset=utf-8, content-length=47, server=Armeria/1.26.4, date=Tue, 5 Mar 2024 04:45:38 GMT], content=CompletableRpcResponse{cause=io.grpc.StatusRuntimeException: CANCELLED}}
io.grpc.StatusRuntimeException: CANCELLED
	at io.grpc.Status.asRuntimeException(Status.java:537) ~[grpc-api-1.58.0.jar:1.58.0]
	at com.linecorp.armeria.internal.common.grpc.StatusAndMetadata.asRuntimeException(StatusAndMetadata.java:53) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.internal.common.grpc.GrpcLogUtil.rpcResponse(GrpcLogUtil.java:62) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.closeListener(AbstractServerCall.java:281) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.closeListener(AbstractServerCall.java:267) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.doClose(AbstractServerCall.java:241) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.close(AbstractServerCall.java:224) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.internal.server.grpc.AbstractServerCall.close(AbstractServerCall.java:219) ~[armeria-grpc-1.26.4.jar:?]
	at com.linecorp.armeria.server.grpc.FramedGrpcService.lambda$startCall$4(FramedGrpcService.java:326) ~[armeria-grpc-1.26.4.jar:?]
	at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:934) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:911) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
	at com.linecorp.armeria.common.util.UnmodifiableFuture.doComplete(UnmodifiableFuture.java:164) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.internal.common.DefaultCancellationScheduler$CancellationFuture.doComplete(DefaultCancellationScheduler.java:523) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.internal.common.DefaultCancellationScheduler.invokeTask(DefaultCancellationScheduler.java:502) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.internal.common.DefaultCancellationScheduler.finishNow0(DefaultCancellationScheduler.java:333) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.internal.common.DefaultCancellationScheduler.finishNow(DefaultCancellationScheduler.java:317) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.internal.server.DefaultServiceRequestContext.cancel(DefaultServiceRequestContext.java:347) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.server.AggregatingDecodedHttpRequest.abortResponse(AggregatingDecodedHttpRequest.java:190) ~[armeria-1.26.4.jar:?]
	at com.linecorp.armeria.server.Http2RequestDecoder.onRstStreamRead(Http2RequestDecoder.java:413) ~[armeria-1.26.4.jar:?]
	at io.netty.handler.codec.http2.Http2FrameListenerDecorator.onRstStreamRead(Http2FrameListenerDecorator.java:59) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.Http2MaxRstFrameListener.onRstStreamRead(Http2MaxRstFrameListener.java:56) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.Http2FrameListenerDecorator.onRstStreamRead(Http2FrameListenerDecorator.java:59) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder$FrameReadListener.onRstStreamRead(DefaultHttp2ConnectionDecoder.java:442) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.Http2InboundFrameLogger$1.onRstStreamRead(Http2InboundFrameLogger.java:80) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readRstStreamFrame(DefaultHttp2FrameReader.java:509) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DefaultHttp2FrameReader.processPayloadState(DefaultHttp2FrameReader.java:259) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:159) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.Http2InboundFrameLogger.readFrame(Http2InboundFrameLogger.java:41) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder.decodeFrame(DefaultHttp2ConnectionDecoder.java:173) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DecoratingHttp2ConnectionDecoder.decodeFrame(DecoratingHttp2ConnectionDecoder.java:63) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.DecoratingHttp2ConnectionDecoder.decodeFrame(DecoratingHttp2ConnectionDecoder.java:63) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:393) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:453) ~[netty-codec-http2-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:529) ~[netty-codec-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:468) ~[netty-codec-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290) ~[netty-codec-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.logging.LoggingHandler.channelRead(LoggingHandler.java:280) ~[netty-handler-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.handler.flush.FlushConsolidationHandler.channelRead(FlushConsolidationHandler.java:152) ~[netty-handler-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[netty-transport-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.kqueue.AbstractKQueueStreamChannel$KQueueStreamUnsafe.readReady(AbstractKQueueStreamChannel.java:544) ~[netty-transport-classes-kqueue-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.kqueue.AbstractKQueueChannel$AbstractKQueueUnsafe.readReady(AbstractKQueueChannel.java:387) ~[netty-transport-classes-kqueue-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.kqueue.KQueueEventLoop.processReady(KQueueEventLoop.java:218) ~[netty-transport-classes-kqueue-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.channel.kqueue.KQueueEventLoop.run(KQueueEventLoop.java:296) ~[netty-transport-classes-kqueue-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.100.Final.jar:4.1.100.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.100.Final.jar:4.1.100.Final]
	at java.lang.Thread.run(Thread.java:840) ~[?:?]
Caused by: com.linecorp.armeria.common.stream.ClosedStreamException: received a RST_STREAM frame: CANCEL

[jrhee17]

Hi! The stacktrace seems slightly out of sync from the main branch.
Do you mind sharing the armeria version used as well?

[angela278]

We're using 1.26.4

[jrhee17]

Or is there a better way of handling this?

If you're looking to handle exceptions, have you been able to take a look at GrpcServiceBuilder#exceptionHandler for exception handling in the gRPC level?

GrpcService.builder()
        .exceptionHandler(new ExceptionHandler())

shows the exception coming from Armeria so I thought I'd try here first

And yes, I believe you're asking in the right place 👍

[angela278]

I haven't looked into the ExceptionHandler yet, but thank you for bringing it to my attention! I will check to see if it fits our use case.