From ecb9ace865dcb985e631de46bf2731af5a1a31aa Mon Sep 17 00:00:00 2001 From: Rui <1685901819@qq.com> Date: Sun, 2 Aug 2026 13:33:29 +0800 Subject: [PATCH] [ISSUE #10747] Fix InvokeCallback completion order Signed-off-by: Rui <1685901819@qq.com> --- .../remoting/netty/NettyRemotingAbstract.java | 34 +++++--- .../netty/NettyRemotingClientTest.java | 83 ++++++++++++++++++- 2 files changed, 104 insertions(+), 13 deletions(-) diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java index a735f8455d3..5d766a0deb9 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java @@ -677,19 +677,31 @@ public void operationFail(Throwable throwable) { public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis, final InvokeCallback invokeCallback) { invokeImpl(channel, request, timeoutMillis) - .whenComplete((v, t) -> { - if (t == null) { - invokeCallback.operationComplete(v); + .whenComplete((responseFuture, t) -> { + final ResponseFuture callbackFuture; + if (t != null) { + callbackFuture = new ResponseFuture(channel, request.getOpaque(), request, timeoutMillis, + null, null); + callbackFuture.setCause(t); } else { - ResponseFuture responseFuture = new ResponseFuture(channel, request.getOpaque(), request, timeoutMillis, null, null); - responseFuture.setCause(t); - invokeCallback.operationComplete(responseFuture); + callbackFuture = responseFuture; + } + + try { + if (t == null) { + invokeCallback.operationSucceed(responseFuture.getResponseCommand()); + } else { + invokeCallback.operationFail(ExceptionUtils.getRealException(t)); + } + } catch (Throwable e) { + log.warn("execute outcome callback in invokeAsyncImpl, and callback throw", e); + } finally { + try { + invokeCallback.operationComplete(callbackFuture); + } catch (Throwable e) { + log.warn("execute completion callback in invokeAsyncImpl, and callback throw", e); + } } - }) - .thenAccept(responseFuture -> invokeCallback.operationSucceed(responseFuture.getResponseCommand())) - .exceptionally(t -> { - invokeCallback.operationFail(ExceptionUtils.getRealException(t)); - return null; }); } diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java index 456e7ecdd59..37b39691bd9 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientTest.java @@ -39,6 +39,7 @@ import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.Spy; import org.mockito.junit.MockitoJUnitRunner; @@ -50,7 +51,10 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.same; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; @@ -226,6 +230,9 @@ public void operationComplete(ResponseFuture responseFuture) { verify(callback, times(1)).operationSucceed(eq(response)); verify(callback, times(1)).operationComplete(eq(responseFuture)); verify(callback, never()).operationFail(any()); + InOrder callbackOrder = inOrder(callback); + callbackOrder.verify(callback).operationSucceed(eq(response)); + callbackOrder.verify(callback).operationComplete(eq(responseFuture)); verify(rpcHookMock).doBeforeRequest(anyString(), eq(request)); verify(rpcHookMock).doAfterResponse(anyString(), eq(request), eq(response)); @@ -238,7 +245,8 @@ public void testInvokeAsyncFail() { Channel channel = new LocalChannel(); CompletableFuture future = new CompletableFuture<>(); - future.completeExceptionally(new RemotingException(null)); + RemotingException failure = new RemotingException("invoke failure sentinel"); + future.completeExceptionally(failure); doReturn(future).when(remotingClient).invoke0(any(Channel.class), any(RemotingCommand.class), anyLong()); @@ -246,12 +254,83 @@ public void testInvokeAsyncFail() { remotingClient.invokeAsyncImpl(channel, request, 1000, callback); verify(callback, never()).operationSucceed(any()); verify(callback, times(1)).operationComplete(any()); - verify(callback, times(1)).operationFail(any()); + verify(callback, times(1)).operationFail(same(failure)); + InOrder callbackOrder = inOrder(callback); + callbackOrder.verify(callback).operationFail(same(failure)); + callbackOrder.verify(callback).operationComplete(any()); verify(rpcHookMock).doBeforeRequest(anyString(), eq(request)); verify(rpcHookMock, never()).doAfterResponse(anyString(), eq(request), any()); } + @Test + public void testInvokeAsyncCompleteFailureDoesNotInvokeFailureCallback() { + remotingClient.registerRPCHook(rpcHookMock); + Channel channel = new LocalChannel(); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null); + RemotingCommand response = RemotingCommand.createResponseCommand(null); + response.setCode(ResponseCode.SUCCESS); + ResponseFuture responseFuture = new ResponseFuture(channel, request.getOpaque(), request, 1000, + responseFuture1 -> { + }, new SemaphoreReleaseOnlyOnce(new Semaphore(1))); + responseFuture.setResponseCommand(response); + CompletableFuture future = CompletableFuture.completedFuture(responseFuture); + doReturn(future).when(remotingClient).invoke0(any(Channel.class), any(RemotingCommand.class), anyLong()); + + RuntimeException sentinel = new RuntimeException("operationComplete sentinel"); + InvokeCallback callback = mock(InvokeCallback.class); + doThrow(sentinel).when(callback).operationComplete(responseFuture); + + remotingClient.invokeAsyncImpl(channel, request, 1000, callback); + + verify(callback).operationSucceed(response); + verify(callback).operationComplete(responseFuture); + verify(callback, never()).operationFail(any()); + } + + @Test + public void testInvokeAsyncSuccessCallbackFailureStillCompletes() { + Channel channel = new LocalChannel(); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null); + RemotingCommand response = RemotingCommand.createResponseCommand(null); + ResponseFuture responseFuture = new ResponseFuture(channel, request.getOpaque(), request, 1000, + responseFuture1 -> { + }, new SemaphoreReleaseOnlyOnce(new Semaphore(1))); + responseFuture.setResponseCommand(response); + CompletableFuture future = CompletableFuture.completedFuture(responseFuture); + doReturn(future).when(remotingClient).invoke0(any(Channel.class), any(RemotingCommand.class), anyLong()); + + InvokeCallback callback = mock(InvokeCallback.class); + doThrow(new RuntimeException("operationSucceed sentinel")).when(callback).operationSucceed(response); + + remotingClient.invokeAsyncImpl(channel, request, 1000, callback); + + InOrder callbackOrder = inOrder(callback); + callbackOrder.verify(callback).operationSucceed(response); + callbackOrder.verify(callback).operationComplete(responseFuture); + verify(callback, never()).operationFail(any()); + } + + @Test + public void testInvokeAsyncFailureCallbackFailureStillCompletes() { + Channel channel = new LocalChannel(); + RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null); + RemotingException failure = new RemotingException("invoke failure sentinel"); + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(failure); + doReturn(future).when(remotingClient).invoke0(any(Channel.class), any(RemotingCommand.class), anyLong()); + + InvokeCallback callback = mock(InvokeCallback.class); + doThrow(new RuntimeException("operationFail callback sentinel")).when(callback).operationFail(failure); + + remotingClient.invokeAsyncImpl(channel, request, 1000, callback); + + InOrder callbackOrder = inOrder(callback); + callbackOrder.verify(callback).operationFail(failure); + callbackOrder.verify(callback).operationComplete(any(ResponseFuture.class)); + verify(callback, never()).operationSucceed(any()); + } + @Test public void testInvokeImpl() throws ExecutionException, InterruptedException { remotingClient.registerRPCHook(rpcHookMock);