You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2022/09/01 01:39:20 UTC
[rocketmq] branch develop updated: [ISSUE #4953] return INVALID_RECEIPT_HANDLE when cannot find the broker of handle (#4954)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui 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 f77377f36 [ISSUE #4953] return INVALID_RECEIPT_HANDLE when cannot find the broker of handle (#4954)
f77377f36 is described below
commit f77377f3644ffa411c4eb1fd4bfe294de43fe237
Author: lk <ka...@alibaba-inc.com>
AuthorDate: Thu Sep 1 09:39:13 2022 +0800
[ISSUE #4953] return INVALID_RECEIPT_HANDLE when cannot find the broker of handle (#4954)
---
.../service/message/ClusterMessageService.java | 16 +++--
.../service/message/ClusterMessageServiceTest.java | 78 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 6 deletions(-)
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java
index e5b7e9533..877cfd43b 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java
@@ -79,7 +79,7 @@ public class ClusterMessageService implements MessageService {
public CompletableFuture<RemotingCommand> sendMessageBack(ProxyContext ctx, ReceiptHandle handle, String messageId,
ConsumerSendMsgBackRequestHeader requestHeader, long timeoutMillis) {
return this.mqClientAPIFactory.getClient().sendMessageBackAsync(
- this.resolveBrokerAddr(handle),
+ this.resolveBrokerAddrInReceiptHandle(handle),
requestHeader,
timeoutMillis
);
@@ -118,7 +118,7 @@ public class ClusterMessageService implements MessageService {
public CompletableFuture<AckResult> changeInvisibleTime(ProxyContext ctx, ReceiptHandle handle, String messageId,
ChangeInvisibleTimeRequestHeader requestHeader, long timeoutMillis) {
return this.mqClientAPIFactory.getClient().changeInvisibleTimeAsync(
- this.resolveBrokerAddr(handle),
+ this.resolveBrokerAddrInReceiptHandle(handle),
handle.getBrokerName(),
requestHeader,
timeoutMillis
@@ -129,7 +129,7 @@ public class ClusterMessageService implements MessageService {
public CompletableFuture<AckResult> ackMessage(ProxyContext ctx, ReceiptHandle handle, String messageId,
AckMessageRequestHeader requestHeader, long timeoutMillis) {
return this.mqClientAPIFactory.getClient().ackMessageAsync(
- this.resolveBrokerAddr(handle),
+ this.resolveBrokerAddrInReceiptHandle(handle),
requestHeader,
timeoutMillis
);
@@ -205,15 +205,19 @@ public class ClusterMessageService implements MessageService {
);
}
- protected String resolveBrokerAddr(ReceiptHandle handle) {
- return resolveBrokerAddr(handle.getBrokerName());
+ protected String resolveBrokerAddrInReceiptHandle(ReceiptHandle handle) {
+ try {
+ return this.topicRouteService.getBrokerAddr(handle.getBrokerName());
+ } catch (Throwable t) {
+ throw new ProxyException(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, "cannot find broker " + handle.getBrokerName(), t);
+ }
}
protected String resolveBrokerAddr(String brokerName) {
try {
return this.topicRouteService.getBrokerAddr(brokerName);
} catch (Throwable t) {
- throw new ProxyException(ProxyExceptionCode.INVALID_BROKER_NAME, "broker " + brokerName + " cannot find", t);
+ throw new ProxyException(ProxyExceptionCode.INVALID_BROKER_NAME, "cannot find broker " + brokerName, t);
}
}
}
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java
new file mode 100644
index 000000000..d564ae10d
--- /dev/null
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.service.message;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.consumer.ReceiptHandle;
+import org.apache.rocketmq.common.message.MessageClientIDSetter;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.common.ProxyException;
+import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
+import org.apache.rocketmq.proxy.service.mqclient.MQClientAPIFactory;
+import org.apache.rocketmq.proxy.service.route.TopicRouteService;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ClusterMessageServiceTest {
+
+ private TopicRouteService topicRouteService;
+ private ClusterMessageService clusterMessageService;
+
+ @Before
+ public void before() {
+ this.topicRouteService = mock(TopicRouteService.class);
+ MQClientAPIFactory mqClientAPIFactory = mock(MQClientAPIFactory.class);
+ this.clusterMessageService = new ClusterMessageService(this.topicRouteService, mqClientAPIFactory);
+ }
+
+ @Test
+ public void testAckMessageByInvalidBrokerNameHandle() throws Exception {
+ when(topicRouteService.getBrokerAddr(anyString())).thenThrow(new MQClientException(ResponseCode.TOPIC_NOT_EXIST, ""));
+ try {
+ this.clusterMessageService.ackMessage(
+ ProxyContext.create(),
+ ReceiptHandle.builder()
+ .startOffset(0L)
+ .retrieveTime(System.currentTimeMillis())
+ .invisibleTime(3000)
+ .reviveQueueId(1)
+ .topicType(ReceiptHandle.NORMAL_TOPIC)
+ .brokerName("notExistBroker")
+ .queueId(0)
+ .offset(123)
+ .commitLogOffset(0L)
+ .build(),
+ MessageClientIDSetter.createUniqID(),
+ new AckMessageRequestHeader(),
+ 3000);
+ fail();
+ } catch (Exception e) {
+ assertTrue(e instanceof ProxyException);
+ ProxyException proxyException = (ProxyException) e;
+ assertEquals(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, proxyException.getCode());
+ }
+ }
+}
\ No newline at end of file