Use the correct error when reset a stream (#13309)

Motivation:

At the moment we always used "CANCEL" when reseting a stream because of an error. This is not correct, we should use the error that caused the stream error.

Modifications:

- Use correct error when reset a stream.
- Add unit test

Result:

Correctly propagate stream errors via reset

codec-http2/src/main/java/io/netty/handler/codec/http2/AbstractHttp2StreamChannel.java

@@ -576,6 +576,11 @@ void fireChildReadComplete() {
         unsafe.notifyReadComplete(unsafe.recvBufAllocHandle(), false);
     }
 
+    final void closeWithError(Http2Error error) {
+        assert eventLoop().inEventLoop();
+        unsafe.close(unsafe.voidPromise(), error);
+    }
+
     private final class Http2ChannelUnsafe implements Unsafe {
         private final VoidChannelPromise unsafeVoidPromise =
                 new VoidChannelPromise(AbstractHttp2StreamChannel.this, false);
@@ -648,6 +653,10 @@ public void disconnect(ChannelPromise promise) {
 
         @Override
         public void close(final ChannelPromise promise) {
+            close(promise, Http2Error.CANCEL);
+        }
+
+        void close(final ChannelPromise promise, Http2Error error) {
             if (!promise.setUncancellable()) {
                 return;
             }
@@ -678,7 +687,7 @@ public void operationComplete(ChannelFuture future) {
             // Only ever send a reset frame if the connection is still alive and if the stream was created before
             // as otherwise we may send a RST on a stream in an invalid state and cause a connection error.
             if (parent().isActive() && !readEOS && isStreamIdValid(stream.id())) {
-                Http2StreamFrame resetFrame = new DefaultHttp2ResetFrame(Http2Error.CANCEL).stream(stream());
+                Http2StreamFrame resetFrame = new DefaultHttp2ResetFrame(error).stream(stream());
                 write(resetFrame, unsafe().voidPromise());
                 flush();
             }

codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexCodec.java

@@ -208,7 +208,9 @@ final void onHttp2FrameStreamException(ChannelHandlerContext ctx, Http2FrameStre
         try {
             channel.pipeline().fireExceptionCaught(cause.getCause());
         } finally {
-            channel.unsafe().closeForcibly();
+            // Close with the correct error that causes this stream exception.
+            // See https://github.com/netty/netty/issues/13235#issuecomment-1441994672
+            channel.closeWithError(cause.error());
         }
     }
 

codec-http2/src/main/java/io/netty/handler/codec/http2/Http2MultiplexHandler.java

@@ -274,7 +274,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, final Throwable cause) th
             try {
                 childChannel.pipeline().fireExceptionCaught(cause.getCause());
             } finally {
-                childChannel.unsafe().closeForcibly();
+                // Close with the correct error that causes this stream exception.
+                // See https://github.com/netty/netty/issues/13235#issuecomment-1441994672
+                childChannel.closeWithError(exception.error());
             }
             return;
         }

codec-http2/src/test/java/io/netty/handler/codec/http2/Http2MultiplexTest.java

@@ -356,6 +356,24 @@ public void execute() throws Throwable {
         assertFalse(channel.isActive());
     }
 
+    @Test
+    public void contentLengthNotMatchRstStreamWithProtocolError() {
+        final LastInboundHandler inboundHandler = new LastInboundHandler();
+        request.addLong(HttpHeaderNames.CONTENT_LENGTH, 10);
+        Http2StreamChannel channel = newInboundStream(3, false, inboundHandler);
+        frameInboundWriter.writeInboundData(3, bb(8), 0, true);
+        assertThrows(StreamException.class, new Executable() {
+            @Override
+            public void execute() throws Throwable {
+                inboundHandler.checkException();
+            }
+        });
+        assertNotNull(inboundHandler.readInbound());
+        assertFalse(channel.isActive());
+        verify(frameWriter).writeRstStream(eqCodecCtx(), eq(3),
+                eq(Http2Error.PROTOCOL_ERROR.code()), anyChannelPromise());
+    }
+
     @Test
     public void framesShouldBeMultiplexed() {
         LastInboundHandler handler1 = new LastInboundHandler();