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/netty5/handler/codec/http2/DefaultHttp2StreamChannel.java

@@ -439,6 +439,11 @@ void fireChildReadComplete() {
         notifyReadComplete(readHandle(), false);
     }
 
+    void closeWithError(Http2Error error) {
+        assert executor().inEventLoop();
+        closeTransport(error, newPromise());
+    }
+
     private void connectTransport(final SocketAddress remoteAddress,
                         SocketAddress localAddress, Promise<Void> promise) {
         if (!promise.setUncancellable()) {
@@ -487,7 +492,11 @@ private void disconnectTransport(Promise<Void> promise) {
         closeTransport(promise);
     }
 
-    private void closeTransport(final Promise<Void> promise) {
+    private void closeTransport(Promise<Void> promise) {
+        closeTransport(Http2Error.CANCEL, promise);
+    }
+
+    private void closeTransport(Http2Error error, final Promise<Void> promise) {
         if (!promise.setUncancellable()) {
             return;
         }
@@ -513,7 +522,7 @@ private void closeTransport(final Promise<Void> promise) {
         // 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());
             writeTransport(resetFrame, newPromise());
             flush();
         }

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

@@ -274,7 +274,9 @@ public void channelExceptionCaught(ChannelHandlerContext ctx, final Throwable ca
             try {
                 childChannel.pipeline().fireChannelExceptionCaught(cause.getCause());
             } finally {
-                childChannel.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/netty5/handler/codec/http2/Http2MultiplexTest.java

@@ -386,6 +386,24 @@ public void execute() throws Throwable {
         assertFalse(channel.isActive());
     }
 
+    @Test
+    public void contentLengthNotMatchRstStreamWithProtocolError() throws Exception {
+        final LastInboundHandler inboundHandler = new LastInboundHandler();
+        request.add(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(any(ChannelHandlerContext.class), eq(3),
+                eq(Http2Error.PROTOCOL_ERROR.code()));
+    }
+
     @Test
     public void framesShouldBeMultiplexed() throws Exception {
         LastInboundHandler handler1 = new LastInboundHandler();