Copilot commented on code in PR #16418:
URL: https://github.com/apache/dubbo/pull/16418#discussion_r3763126275
##########
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStreamTest.java:
##########
@@ -131,4 +132,43 @@ void progress() {
transportListener.onData(buf, false);
Assertions.assertEquals(1, listener.message.length);
}
+
+ @Test
+ void testOnDataReleaseByteBufAfterCallbackExecutorShutdown() {
+ final URL url = URL.valueOf("tri://127.0.0.1:8080/foo.bar.service");
+ final ModuleServiceRepository repo =
+
ApplicationModel.defaultModel().getDefaultModule().getServiceRepository();
+ repo.registerService(IGreeter.class);
+ final ServiceDescriptor serviceDescriptor =
repo.getService(IGreeter.class.getName());
+ final MethodDescriptor methodDescriptor =
serviceDescriptor.getMethod("echo", new Class<?>[] {String.class});
+
+ MockClientStreamListener listener = new MockClientStreamListener();
+ TripleWriteQueue writeQueue = mock(TripleWriteQueue.class);
+ final EmbeddedChannel channel = new EmbeddedChannel();
+ when(writeQueue.enqueueFuture(any(QueuedCommand.class),
any(Executor.class)))
+ .thenReturn(channel.newPromise());
+ Http2StreamChannel http2StreamChannel = mock(Http2StreamChannel.class);
+ when(http2StreamChannel.isActive()).thenReturn(true);
+
when(http2StreamChannel.newSucceededFuture()).thenReturn(channel.newSucceededFuture());
+ when(http2StreamChannel.eventLoop()).thenReturn(new
NioEventLoopGroup().next());
+ when(http2StreamChannel.newPromise()).thenReturn(channel.newPromise());
+ when(http2StreamChannel.parent()).thenReturn(channel);
Review Comment:
The test creates a new NioEventLoopGroup just to return a single EventLoop
from the mocked Http2StreamChannel. This starts threads that are never shut
down, which can leak resources and slow the test suite. You can reuse the
EmbeddedChannel's eventLoop instead.
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractTripleClientStream.java:
##########
@@ -565,14 +565,23 @@ public void onData(ByteBuf data, boolean endStream) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("endStream: {} DATA: {}", endStream,
data.toString(StandardCharsets.UTF_8));
}
+ if (isCallbackExecutorShutdown()) {
+ // The callback executor (e.g. ThreadlessExecutor) has been
shut down, e.g.
+ // after the request timed out {@link AsyncRpcResult}.
SerializingExecutor would
+ // silently drop the submitted task so doOnData would never
run; release the
+ // ByteBuf now to avoid out of heap memory leakage.
+ ReferenceCountUtil.release(data);
+ LOGGER.warn(PROTOCOL_FAILED_RESPONSE, "", "", "Drop late
response data, callback executor is shutdown");
+ return;
Review Comment:
The comment and log message have a few grammar issues and the phrase "out of
heap memory leakage" is inaccurate for Netty ByteBufs (often off-heap). This is
user-visible in logs and can be confusing during incident triage.
This issue also appears on line 580 of the same file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
