Repository: apache/dubbo
File: dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractTripleClientStream.java (Java)

Code before:
            } catch (Throwable t) {
                // Tasks will be rejected when the thread pool is closed or full,
                // ByteBuf needs to be released to avoid out of heap memory leakage.
                // For example, ThreadLessExecutor will be shutdown when request timeout {@link AsyncRpcResult}
                ReferenceCountUtil.release(data);
                LOGGER.error(PROTOCOL_FAILED_RESPONSE, "", "", "submit onData task failed", t);
            }
        }

        private void doOnData(ByteBuf data, boolean endStream) {
            if (transportError != null) {
                transportError.appendDescription("Data:" + data.toString(StandardCharsets.UTF_8));
                ReferenceCountUtil.release(data);
                if (transportError.description.length() > 512 || endStream) {
                    handleH2TransportError(transportError);
                }
                return;
            }
            if (!headerReceived) {
                handleH2TransportError(TriRpcStatus.INTERNAL.withDescription("headers not received before payload"));
                return;
            }
            // Use ByteBufInputStream to adapt ByteBuf to InputStream
            // The second parameter 'true' means release ByteBuf after reading
            deframer.decode(new ByteBufInputStream(data, true));
        }

        @Override
        public void cancelByRemote(long errorCode) {
            executor.execute(() -> {
                transportError =
                        TriRpcStatus.CANCELLED.withDescription("Canceled by remote peer, errorCode=" + errorCode);
                finishProcess(transportError, null, false);
            });
        }

        @Override
        public void onClose() {
            executor.execute(listener::onClose);
        }

        @Override
        public void onWritabilityChanged() {
            AbstractTripleClientStream.this.onWritabilityChanged();
        }
    }
}


Review comment (suggestion):
This ByteBufInputStream is not always closed on method exit.
```suggestion
            try (InputStream in = new ByteBufInputStream(data, true)) {
                deframer.decode(in);
            }
```

Code after:
            } catch (Throwable t) {
                // Tasks will be rejected when the thread pool is closed or full,
                // ByteBuf needs to be released to avoid out of heap memory leakage.
                // For example, ThreadLessExecutor will be shutdown when request timeout {@link AsyncRpcResult}
                ReferenceCountUtil.release(data);
                LOGGER.error(PROTOCOL_FAILED_RESPONSE, "", "", "submit onData task failed", t);
            }
        }

        private void doOnData(ByteBuf data, boolean endStream) {
            if (transportError != null) {
                transportError.appendDescription("Data:" + data.toString(StandardCharsets.UTF_8));
                ReferenceCountUtil.release(data);
                if (transportError.description.length() > 512 || endStream) {
                    handleH2TransportError(transportError);
                }
                return;
            }
            if (!headerReceived) {
                ReferenceCountUtil.release(data);
                handleH2TransportError(TriRpcStatus.INTERNAL.withDescription("headers not received before payload"));
                return;
            }
            // Use ByteBufInputStream to adapt ByteBuf to InputStream
            // The second parameter 'true' means release ByteBuf after reading
            deframer.decode(new ByteBufInputStream(data, true));
        }

        @Override
        public void cancelByRemote(long errorCode) {
            executor.execute(() -> {
                transportError =
                        TriRpcStatus.CANCELLED.withDescription("Canceled by remote peer, errorCode=" + errorCode);
                finishProcess(transportError, null, false);
            });
        }

        @Override
        public void onClose() {
            executor.execute(listener::onClose);
        }

        @Override
        public void onWritabilityChanged() {
            AbstractTripleClientStream.this.onWritabilityChanged();
        }
    }
}