This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 1251c9af3e [ISSUE #10280] Restore opaque when proxy remoting forward
fails (#10281)
1251c9af3e is described below
commit 1251c9af3ef7942178b70c8aad9b04d46cddbb39
Author: yx9o <[email protected]>
AuthorDate: Sat May 23 17:09:38 2026 +0800
[ISSUE #10280] Restore opaque when proxy remoting forward fails (#10281)
---
.../proxy/processor/DefaultMessagingProcessor.java | 21 ++--
.../processor/DefaultMessagingProcessorTest.java | 110 +++++++++++++++++++++
2 files changed, 124 insertions(+), 7 deletions(-)
diff --git
a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
index 274cbf37da..3e7a889485 100644
---
a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
+++
b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
@@ -23,6 +23,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.AclUtils;
@@ -304,10 +305,8 @@ public class DefaultMessagingProcessor extends
AbstractStartAndShutdown implemen
long timeoutMillis) {
int originalRequestOpaque = request.getOpaque();
request.setOpaque(RemotingCommand.createNewRequestId());
- return this.requestBrokerProcessor.request(ctx, brokerName, request,
timeoutMillis).thenApply(r -> {
- request.setOpaque(originalRequestOpaque);
- return r;
- });
+ return restoreRequestOpaque(request, originalRequestOpaque,
+ () -> this.requestBrokerProcessor.request(ctx, brokerName,
request, timeoutMillis));
}
@Override
@@ -315,10 +314,18 @@ public class DefaultMessagingProcessor extends
AbstractStartAndShutdown implemen
long timeoutMillis) {
int originalRequestOpaque = request.getOpaque();
request.setOpaque(RemotingCommand.createNewRequestId());
- return this.requestBrokerProcessor.requestOneway(ctx, brokerName,
request, timeoutMillis).thenApply(r -> {
+ return restoreRequestOpaque(request, originalRequestOpaque,
+ () -> this.requestBrokerProcessor.requestOneway(ctx, brokerName,
request, timeoutMillis));
+ }
+
+ private <T> CompletableFuture<T> restoreRequestOpaque(RemotingCommand
request, int originalRequestOpaque,
+ Supplier<CompletableFuture<T>> requestFutureSupplier) {
+ try {
+ return requestFutureSupplier.get().whenComplete((r, t) ->
request.setOpaque(originalRequestOpaque));
+ } catch (RuntimeException t) {
request.setOpaque(originalRequestOpaque);
- return r;
- });
+ throw t;
+ }
}
@Override
diff --git
a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessorTest.java
b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessorTest.java
new file mode 100644
index 0000000000..9a3ea987d0
--- /dev/null
+++
b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessorTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.processor;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.when;
+
+public class DefaultMessagingProcessorTest extends BaseProcessorTest {
+
+ private DefaultMessagingProcessor defaultMessagingProcessor;
+
+ @Before
+ public void before() throws Throwable {
+ super.before();
+ this.defaultMessagingProcessor = new
DefaultMessagingProcessor(this.serviceManager);
+ }
+
+ @Test
+ public void testRequestShouldRestoreOpaqueWhenForwardFails() {
+ CompletableFuture<RemotingCommand> forwardFuture = new
CompletableFuture<>();
+ forwardFuture.completeExceptionally(new RuntimeException("mock forward
failed"));
+ AtomicInteger forwardOpaque = new AtomicInteger(-1);
+ when(this.messageService.request(any(), anyString(), any(),
anyLong())).thenAnswer(invocation -> {
+ forwardOpaque.set(((RemotingCommand)
invocation.getArgument(2)).getOpaque());
+ return forwardFuture;
+ });
+
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);
+ int originalOpaque = 12345;
+ request.setOpaque(originalOpaque);
+
+ Assert.assertThrows(CompletionException.class,
+ () -> this.defaultMessagingProcessor.request(createContext(),
"broker-a", request, 3000).join());
+
+ Assert.assertNotEquals(originalOpaque, forwardOpaque.get());
+ Assert.assertEquals(originalOpaque, request.getOpaque());
+ }
+
+ @Test
+ public void testRequestOnewayShouldRestoreOpaqueWhenForwardFails() {
+ CompletableFuture<Void> forwardFuture = new CompletableFuture<>();
+ forwardFuture.completeExceptionally(new RuntimeException("mock oneway
forward failed"));
+ when(this.messageService.requestOneway(any(), anyString(), any(),
anyLong())).thenReturn(forwardFuture);
+
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);
+ int originalOpaque = 12345;
+ request.setOpaque(originalOpaque);
+
+ Assert.assertThrows(CompletionException.class,
+ () ->
this.defaultMessagingProcessor.requestOneway(createContext(), "broker-a",
request, 3000).join());
+
+ Assert.assertEquals(originalOpaque, request.getOpaque());
+ }
+
+ @Test
+ public void testRequestShouldRestoreOpaqueWhenForwardThrows() {
+ when(this.messageService.request(any(), anyString(), any(), anyLong()))
+ .thenThrow(new RuntimeException("mock forward throws"));
+
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);
+ int originalOpaque = 12345;
+ request.setOpaque(originalOpaque);
+
+ Assert.assertThrows(RuntimeException.class,
+ () -> this.defaultMessagingProcessor.request(createContext(),
"broker-a", request, 3000));
+
+ Assert.assertEquals(originalOpaque, request.getOpaque());
+ }
+
+ @Test
+ public void testRequestOnewayShouldRestoreOpaqueWhenForwardThrows() {
+ when(this.messageService.requestOneway(any(), anyString(), any(),
anyLong()))
+ .thenThrow(new RuntimeException("mock oneway forward throws"));
+
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);
+ int originalOpaque = 12345;
+ request.setOpaque(originalOpaque);
+
+ Assert.assertThrows(RuntimeException.class,
+ () ->
this.defaultMessagingProcessor.requestOneway(createContext(), "broker-a",
request, 3000));
+
+ Assert.assertEquals(originalOpaque, request.getOpaque());
+ }
+}
