Log expected STREAM_CLOSED exceptions for already closed streams at DEBUG level (#9798)

Motivation:

There is an intrinsic race between a local session resetting a stream
and the peer no longer sending any frames. This can result in the
session receiving frames for a stream that the local peer no longer
tracks. This results in a StreamException being thrown which triggers a
RST_STREAM frame, which is a good thing, but also logging at level WARN,
which is noisy for an expected and benign condition.

Modification:

Change the log level to DEBUG when logging stream errors with code
STREAM_CLOSED. All others are more interesting and will continue to be
logged at level WARN.

Additionally, it was found that DATA frames for streams that could not
have existed only resulted in a StreamException when the spec is clear
that such a situation should be fatal to the connection, resulting in a
GOAWAY(PROTOCOL_ERROR).

Fixes #8025.

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

@@ -597,6 +597,11 @@ public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
                             ctx.channel(), frameName, streamId);
                     return true;
                 }
+
+                // Make sure it's not an out-of-order frame, like a rogue DATA frame, for a stream that could
+                // never have existed.
+                verifyStreamMayHaveExisted(streamId);
+
                 // Its possible that this frame would result in stream ID out of order creation (PROTOCOL ERROR) and its
                 // also possible that this frame is received on a CLOSED stream (STREAM_CLOSED after a RST_STREAM is
                 // sent). We don't have enough information to know for sure, so we choose the lesser of the two errors.

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

@@ -32,6 +32,7 @@ import io.netty.util.collection.IntObjectHashMap;
 import io.netty.util.collection.IntObjectMap;
 import io.netty.util.concurrent.Future;
 import io.netty.util.internal.UnstableApi;
+import io.netty.util.internal.logging.InternalLogLevel;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -523,8 +524,12 @@ public class Http2FrameCodec extends Http2ConnectionHandler {
 
     private void onHttp2UnknownStreamError(@SuppressWarnings("unused") ChannelHandlerContext ctx, Throwable cause,
                                    Http2Exception.StreamException streamException) {
-        // Just log....
-        LOG.warn("Stream exception thrown for unknown stream {}.", streamException.streamId(), cause);
+        // It is normal to hit a race condition where we still receive frames for a stream that this
+        // peer has deemed closed, such as if this peer sends a RST(CANCEL) to discard the request.
+        // Since this is likely to be normal we log at DEBUG level.
+        InternalLogLevel level =
+            streamException.error() == Http2Error.STREAM_CLOSED ? InternalLogLevel.DEBUG : InternalLogLevel.WARN;
+        LOG.log(level, "Stream exception thrown for unknown stream {}.", streamException.streamId(), cause);
     }
 
     @Override

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

@@ -45,7 +45,11 @@ import static io.netty.handler.codec.http2.Http2Stream.State.OPEN;
 import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_REMOTE;
 import static io.netty.handler.codec.http2.Http2TestUtil.newHttp2HeadersWithRequestPseudoHeaders;
 import static io.netty.util.CharsetUtil.UTF_8;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.not;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.any;
@@ -239,9 +243,9 @@ public class DefaultHttp2ConnectionDecoderTest {
         }
     }
 
-    @Test(expected = Http2Exception.class)
+    @Test(expected = Http2Exception.StreamException.class)
     public void dataReadForUnknownStreamShouldApplyFlowControlAndFail() throws Exception {
-        when(connection.streamMayHaveExisted(STREAM_ID)).thenReturn(false);
+        when(connection.streamMayHaveExisted(STREAM_ID)).thenReturn(true);
         when(connection.stream(STREAM_ID)).thenReturn(null);
         final ByteBuf data = dummyData();
         int padding = 10;
@@ -262,6 +266,32 @@ public class DefaultHttp2ConnectionDecoderTest {
         }
     }
 
+    @Test(expected = Http2Exception.class)
+    public void dataReadForUnknownStreamThatCouldntExistFail() throws Exception {
+        when(connection.streamMayHaveExisted(STREAM_ID)).thenReturn(false);
+        when(connection.stream(STREAM_ID)).thenReturn(null);
+        final ByteBuf data = dummyData();
+        int padding = 10;
+        int processedBytes = data.readableBytes() + padding;
+        try {
+            decode().onDataRead(ctx, STREAM_ID, data, padding, true);
+        } catch (Http2Exception ex) {
+            assertThat(ex, not(instanceOf(Http2Exception.StreamException.class)));
+            throw ex;
+        } finally {
+            try {
+                verify(localFlow)
+                    .receiveFlowControlledFrame(eq((Http2Stream) null), eq(data), eq(padding), eq(true));
+                verify(localFlow).consumeBytes(eq((Http2Stream) null), eq(processedBytes));
+                verify(localFlow).frameWriter(any(Http2FrameWriter.class));
+                verifyNoMoreInteractions(localFlow);
+                verify(listener, never()).onDataRead(eq(ctx), anyInt(), any(ByteBuf.class), anyInt(), anyBoolean());
+            } finally {
+                data.release();
+            }
+        }
+    }
+
     @Test
     public void dataReadForUnknownStreamShouldApplyFlowControl() throws Exception {
         when(connection.stream(STREAM_ID)).thenReturn(null);