From 487fa4f7c73a9339e69409c934e4ec669ba084e1 Mon Sep 17 00:00:00 2001
From: TomlongTK <longqiang02@maoyan.com>
Date: Wed, 8 May 2024 14:08:57 +0800
Subject: [PATCH] Fix memory leak (#14127)

---
 .../AbstractServerHttpChannelObserver.java       |  9 +++++++--
 .../dubbo/remoting/http12/HttpInputMessage.java  |  8 +++++++-
 .../dubbo/remoting/http12/HttpOutputMessage.java |  8 +++++++-
 .../remoting/http12/h1/DefaultHttp1Request.java  |  6 ++++++
 .../remoting/http12/h1/DefaultHttp1Response.java |  6 ++++++
 .../remoting/http12/h1/Http1OutputMessage.java   | 11 +++++++++++
 .../http12/h2/Http2OutputMessageFrame.java       | 11 +++++++++++
 .../tri/h12/AbstractServerTransportListener.java | 16 +++++++++++++++-
 .../grpc/GrpcHttp2ServerTransportListener.java   | 14 ++++++++++++++
 .../GenericHttp2ServerTransportListener.java     |  3 ++-
 10 files changed, 86 insertions(+), 6 deletions(-)

diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java
index 515afb2d50..9883a8072e 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java
@@ -162,8 +162,13 @@ public abstract class AbstractServerHttpChannelObserver implements CustomizableH
             data = ((HttpResult<?>) data).getBody();
         }
         HttpOutputMessage outputMessage = encodeHttpOutputMessage(data);
-        preOutputMessage(outputMessage);
-        responseEncoder.encode(outputMessage.getBody(), data);
+        try {
+            preOutputMessage(outputMessage);
+            responseEncoder.encode(outputMessage.getBody(), data);
+        } catch (Throwable t) {
+            outputMessage.close();
+            throw t;
+        }
         return outputMessage;
     }
 
diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpInputMessage.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpInputMessage.java
index 60d6a4dec7..717b03665a 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpInputMessage.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpInputMessage.java
@@ -16,9 +16,15 @@
  */
 package org.apache.dubbo.remoting.http12;
 
+import java.io.IOException;
 import java.io.InputStream;
 
-public interface HttpInputMessage {
+public interface HttpInputMessage extends AutoCloseable {
 
     InputStream getBody();
+
+    @Override
+    default void close() throws IOException {
+        getBody().close();
+    }
 }
diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpOutputMessage.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpOutputMessage.java
index e56630430c..ac64685bd0 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpOutputMessage.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpOutputMessage.java
@@ -17,9 +17,10 @@
 package org.apache.dubbo.remoting.http12;
 
 import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.OutputStream;
 
-public interface HttpOutputMessage {
+public interface HttpOutputMessage extends AutoCloseable {
 
     HttpOutputMessage EMPTY_MESSAGE = new HttpOutputMessage() {
 
@@ -32,4 +33,9 @@ public interface HttpOutputMessage {
     };
 
     OutputStream getBody();
+
+    @Override
+    default void close() throws IOException {
+        getBody().close();
+    }
 }
diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Request.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Request.java
index 1bc610d84b..d950f8a040 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Request.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Request.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.remoting.http12.HttpHeaders;
 import org.apache.dubbo.remoting.http12.HttpInputMessage;
 import org.apache.dubbo.remoting.http12.RequestMetadata;
 
+import java.io.IOException;
 import java.io.InputStream;
 
 public class DefaultHttp1Request implements Http1Request {
@@ -52,4 +53,9 @@ public class DefaultHttp1Request implements Http1Request {
     public String path() {
         return httpMetadata.path();
     }
+
+    @Override
+    public void close() throws IOException {
+        httpInputMessage.close();
+    }
 }
diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Response.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Response.java
index 4f345b6cbe..75bdb4210c 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Response.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Response.java
@@ -20,6 +20,7 @@ import org.apache.dubbo.remoting.http12.HttpHeaders;
 import org.apache.dubbo.remoting.http12.HttpInputMessage;
 import org.apache.dubbo.remoting.http12.HttpMetadata;
 
+import java.io.IOException;
 import java.io.InputStream;
 
 public class DefaultHttp1Response implements HttpMetadata, HttpInputMessage {
@@ -42,4 +43,9 @@ public class DefaultHttp1Response implements HttpMetadata, HttpInputMessage {
     public HttpHeaders headers() {
         return httpMetadata.headers();
     }
+
+    @Override
+    public void close() throws IOException {
+        httpInputMessage.close();
+    }
 }
diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1OutputMessage.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1OutputMessage.java
index 2406b8249c..dd8844162b 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1OutputMessage.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1OutputMessage.java
@@ -18,8 +18,11 @@ package org.apache.dubbo.remoting.http12.h1;
 
 import org.apache.dubbo.remoting.http12.HttpOutputMessage;
 
+import java.io.IOException;
 import java.io.OutputStream;
 
+import io.netty.buffer.ByteBufOutputStream;
+
 public class Http1OutputMessage implements HttpOutputMessage {
 
     private final OutputStream outputStream;
@@ -32,4 +35,12 @@ public class Http1OutputMessage implements HttpOutputMessage {
     public OutputStream getBody() {
         return outputStream;
     }
+
+    @Override
+    public void close() throws IOException {
+        if (outputStream instanceof ByteBufOutputStream) {
+            ((ByteBufOutputStream) outputStream).buffer().release();
+        }
+        outputStream.close();
+    }
 }
diff --git a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessageFrame.java b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessageFrame.java
index 15c87eb413..d3ba8ce66d 100644
--- a/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessageFrame.java
+++ b/dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessageFrame.java
@@ -16,8 +16,11 @@
  */
 package org.apache.dubbo.remoting.http12.h2;
 
+import java.io.IOException;
 import java.io.OutputStream;
 
+import io.netty.buffer.ByteBufOutputStream;
+
 public class Http2OutputMessageFrame implements Http2OutputMessage {
 
     private final OutputStream body;
@@ -42,6 +45,14 @@ public class Http2OutputMessageFrame implements Http2OutputMessage {
         return body;
     }
 
+    @Override
+    public void close() throws IOException {
+        if (body instanceof ByteBufOutputStream) {
+            ((ByteBufOutputStream) body).buffer().release();
+        }
+        body.close();
+    }
+
     @Override
     public boolean isEndStream() {
         return endStream;
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java
index 5d74b831b0..970aa88fa7 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/AbstractServerTransportListener.java
@@ -126,7 +126,9 @@ public abstract class AbstractServerTransportListener<HEADER extends RequestMeta
                 doOnData(message);
             } catch (Throwable t) {
                 logError(t);
-                onError(t);
+                onError(message, t);
+            } finally {
+                onFinally(message);
             }
         });
     }
@@ -184,6 +186,18 @@ public abstract class AbstractServerTransportListener<HEADER extends RequestMeta
         throw new HttpStatusException(HttpStatus.INTERNAL_SERVER_ERROR.getCode(), throwable);
     }
 
+    protected void onError(MESSAGE message, Throwable throwable) {
+        onError(throwable);
+    }
+
+    protected void onFinally(MESSAGE message) {
+        try {
+            message.close();
+        } catch (Exception e) {
+            onError(e);
+        }
+    }
+
     protected RpcInvocation buildRpcInvocation(RpcInvocationBuildContext context) {
         MethodDescriptor methodDescriptor = context.getMethodDescriptor();
         if (methodDescriptor == null) {
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java
index 9ae7881302..0399c64334 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcHttp2ServerTransportListener.java
@@ -26,6 +26,7 @@ import org.apache.dubbo.remoting.http12.exception.DecodeException;
 import org.apache.dubbo.remoting.http12.exception.UnimplementedException;
 import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
 import org.apache.dubbo.remoting.http12.h2.Http2Header;
+import org.apache.dubbo.remoting.http12.h2.Http2InputMessage;
 import org.apache.dubbo.remoting.http12.h2.Http2TransportListener;
 import org.apache.dubbo.remoting.http12.message.MethodMetadata;
 import org.apache.dubbo.remoting.http12.message.StreamingDecoder;
@@ -122,6 +123,19 @@ public class GrpcHttp2ServerTransportListener extends GenericHttp2ServerTranspor
         return invocation;
     }
 
+    @Override
+    protected void onError(Http2InputMessage message, Throwable throwable) {
+        try {
+            message.close();
+        } catch (Exception e) {
+            throwable.addSuppressed(e);
+        }
+        onError(throwable);
+    }
+
+    @Override
+    protected void onFinally(Http2InputMessage message) {}
+
     @Override
     protected GrpcStreamingDecoder getStreamingDecoder() {
         return (GrpcStreamingDecoder) super.getStreamingDecoder();
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java
index 326e09ede3..4fee48dfb4 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/GenericHttp2ServerTransportListener.java
@@ -86,6 +86,7 @@ public class GenericHttp2ServerTransportListener extends AbstractServerTransport
         return new SerializingExecutor(executorSupport.getExecutor(metadata));
     }
 
+    @Override
     protected void doOnMetadata(Http2Header metadata) {
         if (metadata.isEndStream()) {
             if (!HttpMethods.supportBody(metadata.method())) {
@@ -164,7 +165,7 @@ public class GenericHttp2ServerTransportListener extends AbstractServerTransport
     @Override
     protected void onDataCompletion(Http2InputMessage message) {
         if (message.isEndStream()) {
-            serverCallListener.onComplete();
+            getStreamingDecoder().close();
         }
     }