Fix buffer leak regression in HttpClientCodec (#12762)

Motivation:

#12709 changed HttpObjectEncoder to override the write method of MessageToMessageEncoder, with slightly changed semantics: The `msg` argument to `encode` is not released anymore. To accommodate this change, #12709 also update `HttpObjectEncoder.encode` to release the `msg`. However, `HttpClientCodec.Encoder` overrides `encode` and simply forwards the message if a HTTP upgrade has been completed. This code path was not updated to release the input message. This leads to a memory leak.

Modifications:

Changed the `encode` implementation to not retain the message that is forwarded. Added a test case to verify that the refCnt to the data passed through is unchanged.

Result:

The buffer retains its correct refCnt and will be released properly.

codec-http/src/main/java/io/netty/handler/codec/http/HttpClientCodec.java

@@ -191,7 +191,8 @@ protected void encode(
                 ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
 
             if (upgraded) {
-                out.add(ReferenceCountUtil.retain(msg));
+                // HttpObjectEncoder overrides .write and does not release msg, so we don't need to retain it here
+                out.add(msg);
                 return;
             }
 

codec-http/src/test/java/io/netty/handler/codec/http/HttpClientCodecTest.java

@@ -36,6 +36,7 @@
 import io.netty.util.CharsetUtil;
 import io.netty.util.NetUtil;
 import org.hamcrest.CoreMatchers;
+import org.hamcrest.Matchers;
 import org.junit.jupiter.api.Test;
 
 import java.net.InetSocketAddress;
@@ -423,4 +424,19 @@ public void testMultipleResponses() {
         assertTrue(ch.finishAndReleaseAll());
     }
 
+    @Test
+    public void testWriteThroughAfterUpgrade() {
+        HttpClientCodec codec = new HttpClientCodec();
+        EmbeddedChannel ch = new EmbeddedChannel(codec);
+        codec.prepareUpgradeFrom(null);
+
+        ByteBuf buffer = ch.alloc().buffer();
+        assertThat(buffer.refCnt(), is(1));
+        assertTrue(ch.writeOutbound(buffer));
+        // buffer should pass through unchanged
+        assertThat(ch.<ByteBuf>readOutbound(), sameInstance(buffer));
+        assertThat(buffer.refCnt(), is(1));
+
+        buffer.release();
+    }
 }