diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteService.java index a4df98971cb..11b1f92b86e 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteService.java @@ -21,6 +21,8 @@ import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.proxy.common.Address; 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.remoting.protocol.route.TopicRouteData; public class ClusterTopicRouteService extends TopicRouteService { @@ -44,7 +46,8 @@ public ProxyTopicRouteData getTopicRouteForProxy(ProxyContext ctx, List
@Override public String getBrokerAddr(ProxyContext ctx, String brokerName) throws Exception { TopicRouteWrapper topicRouteWrapper = getAllMessageQueueView(ctx, brokerName).getTopicRouteWrapper(); - return topicRouteWrapper.getMasterAddr(brokerName); + return topicRouteWrapper.getOptionalMasterAddr(brokerName) + .orElseThrow(() -> new ProxyException(ProxyExceptionCode.INVALID_BROKER_NAME, "cannot find broker " + brokerName)); } @Override diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java index 8f08c1df0e5..e55b79cf397 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelector.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadLocalRandom; @@ -32,15 +33,20 @@ import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.tuple.Pair; +import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.constant.PermName; import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; import org.apache.rocketmq.remoting.protocol.route.QueueData; import static org.apache.rocketmq.proxy.service.route.MessageQueuePenalizer.selectLeastPenaltyWithPriority; import static org.apache.rocketmq.proxy.service.route.MessageQueuePriorityProvider.buildPriorityGroups; public class MessageQueueSelector { + private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); private static final int BROKER_ACTING_QUEUE_ID = -1; + private static final int MAX_ORDER_TOPIC_WRITE_QUEUE_COUNT = 1024; // multiple queues for brokers with queueId : normal private final List queues = new ArrayList<>(); @@ -110,17 +116,27 @@ private static List buildWrite(TopicRouteWrapper topicR String[] brokers = topicRoute.getOrderTopicConf().split(";"); for (String broker : brokers) { String[] item = broker.split(":"); + if (item.length != 2 || StringUtils.isBlank(item[0]) || StringUtils.isBlank(item[1])) { + log.warn("skip invalid order topic route item. topic:{}, item:{}", topicRoute.getTopicName(), broker); + continue; + } + String brokerName = item[0]; - String brokerAddr = topicRoute.getMasterAddr(brokerName); - if (brokerAddr == null) { + Optional brokerAddr = topicRoute.getOptionalMasterAddr(brokerName); + if (!brokerAddr.isPresent()) { + log.warn("skip order topic route item without master broker address. topic:{}, brokerName:{}", + topicRoute.getTopicName(), brokerName); continue; } - int nums = Integer.parseInt(item[1]); - for (int i = 0; i < nums; i++) { + Optional nums = parseOrderTopicQueueCount(topicRoute.getTopicName(), broker); + if (!nums.isPresent()) { + continue; + } + for (int i = 0; i < nums.get(); i++) { AddressableMessageQueue mq = new AddressableMessageQueue( new MessageQueue(topicRoute.getTopicName(), brokerName, i), - brokerAddr); + brokerAddr.get()); queueSet.add(mq); } } @@ -132,15 +148,15 @@ private static List buildWrite(TopicRouteWrapper topicR for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { - String brokerAddr = topicRoute.getMasterAddr(qd.getBrokerName()); - if (brokerAddr == null) { + Optional brokerAddr = topicRoute.getOptionalMasterAddr(qd.getBrokerName()); + if (!brokerAddr.isPresent()) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { AddressableMessageQueue mq = new AddressableMessageQueue( new MessageQueue(topicRoute.getTopicName(), qd.getBrokerName(), i), - brokerAddr); + brokerAddr.get()); queueSet.add(mq); } } @@ -150,6 +166,22 @@ private static List buildWrite(TopicRouteWrapper topicR return queueSet.stream().sorted().collect(Collectors.toList()); } + private static Optional parseOrderTopicQueueCount(String topicName, String broker) { + String[] item = broker.split(":"); + try { + int queueCount = Integer.parseInt(item[1]); + if (queueCount < 1 || queueCount > MAX_ORDER_TOPIC_WRITE_QUEUE_COUNT) { + log.warn("skip order topic route item with out-of-range queue count. topic:{}, item:{}, min:{}, max:{}", + topicName, broker, 1, MAX_ORDER_TOPIC_WRITE_QUEUE_COUNT); + return Optional.empty(); + } + return Optional.of(queueCount); + } catch (NumberFormatException e) { + log.warn("skip order topic route item with invalid queue count. topic:{}, item:{}", topicName, broker); + return Optional.empty(); + } + } + private void buildBrokerActingQueues(String topic, List normalQueues) { for (AddressableMessageQueue mq : normalQueues) { AddressableMessageQueue brokerActingQueue = new AddressableMessageQueue( diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteWrapper.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteWrapper.java index 7956c6284ea..20679fc7001 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteWrapper.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteWrapper.java @@ -43,7 +43,15 @@ public TopicRouteWrapper(TopicRouteData topicRouteData, String topicName) { } public String getMasterAddr(String brokerName) { - return this.brokerNameRouteData.get(brokerName).getBrokerAddrs().get(MixAll.MASTER_ID); + return getOptionalMasterAddr(brokerName).orElse(null); + } + + public Optional getOptionalMasterAddr(String brokerName) { + BrokerData brokerData = this.brokerNameRouteData.get(brokerName); + if (brokerData == null || brokerData.getBrokerAddrs() == null) { + return Optional.empty(); + } + return Optional.ofNullable(brokerData.getBrokerAddrs().get(MixAll.MASTER_ID)); } public String getMasterAddrPrefer(String brokerName) { 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 index 7e4d25f0c09..b976f97ada2 100644 --- 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 @@ -16,14 +16,17 @@ */ package org.apache.rocketmq.proxy.service.message; +import java.util.concurrent.ExecutionException; + import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.impl.mqclient.MQClientAPIFactory; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.MessageClientIDSetter; 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.client.impl.mqclient.MQClientAPIFactory; import org.apache.rocketmq.proxy.service.route.TopicRouteService; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.header.AckMessageRequestHeader; import org.junit.Before; @@ -76,4 +79,36 @@ public void testAckMessageByInvalidBrokerNameHandle() throws Exception { assertEquals(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, proxyException.getCode()); } } + + @Test + public void testRequestCompletesExceptionallyWhenBrokerNameIsInvalid() throws Exception { + when(topicRouteService.getBrokerAddr(any(), anyString())) + .thenThrow(new ProxyException(ProxyExceptionCode.INVALID_BROKER_NAME, "cannot find broker")); + + try { + this.clusterMessageService.request( + ProxyContext.create(), "notExistBroker", RemotingCommand.createRequestCommand(0, null), 3000).get(); + fail(); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof ProxyException); + ProxyException proxyException = (ProxyException) e.getCause(); + assertEquals(ProxyExceptionCode.INVALID_BROKER_NAME, proxyException.getCode()); + } + } + + @Test + public void testRequestOnewayCompletesExceptionallyWhenBrokerNameIsInvalid() throws Exception { + when(topicRouteService.getBrokerAddr(any(), anyString())) + .thenThrow(new ProxyException(ProxyExceptionCode.INVALID_BROKER_NAME, "cannot find broker")); + + try { + this.clusterMessageService.requestOneway( + ProxyContext.create(), "notExistBroker", RemotingCommand.createRequestCommand(0, null), 3000).get(); + fail(); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof ProxyException); + ProxyException proxyException = (ProxyException) e.getCause(); + assertEquals(ProxyExceptionCode.INVALID_BROKER_NAME, proxyException.getCode()); + } + } } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java index 15d83483b9d..d884896e0c9 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/ClusterTopicRouteServiceTest.java @@ -32,6 +32,8 @@ import org.apache.rocketmq.common.thread.ThreadPoolMonitor; import org.apache.rocketmq.proxy.common.Address; 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.BaseServiceTest; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.route.BrokerData; @@ -58,6 +60,7 @@ public class ClusterTopicRouteServiceTest extends BaseServiceTest { protected static final String BROKER2_NAME = "broker2"; protected static final String BROKER2_ADDR = "127.0.0.2:10911"; + protected static final String UNKNOWN_BROKER_NAME = "unknownBroker"; @Before public void before() throws Throwable { @@ -96,6 +99,7 @@ public void before() throws Throwable { brokerTopicRouteData.setQueueDatas(Lists.newArrayList(queueData, queue2Data)); when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(BROKER_NAME), anyLong())).thenReturn(brokerTopicRouteData); when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(BROKER2_NAME), anyLong())).thenReturn(brokerTopicRouteData); + when(this.mqClientAPIExt.getTopicRouteInfoFromNameServer(eq(UNKNOWN_BROKER_NAME), anyLong())).thenReturn(brokerTopicRouteData); } @Test @@ -116,6 +120,16 @@ public void testGetBrokerAddr() throws Throwable { assertEquals(BROKER2_ADDR, topicRouteService.getBrokerAddr(ctx, BROKER2_NAME)); } + @Test + public void testGetBrokerAddrThrowsForUnknownBroker() { + ProxyContext ctx = ProxyContext.create(); + + ProxyException exception = catchThrowableOfType(() -> + topicRouteService.getBrokerAddr(ctx, UNKNOWN_BROKER_NAME), ProxyException.class); + + assertEquals(ProxyExceptionCode.INVALID_BROKER_NAME, exception.getCode()); + } + @Test public void testGetTopicRouteForProxy() throws Throwable { ProxyContext ctx = ProxyContext.create(); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java index e44ed28f4a6..235b24c63dc 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/route/MessageQueueSelectorTest.java @@ -22,6 +22,8 @@ import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; public class MessageQueueSelectorTest extends BaseServiceTest { @@ -81,4 +83,52 @@ public void testWriteMessageQueue() { messageQueueSelector.selectOne(false); assertEquals(queue, messageQueueSelector.selectOne(false)); } -} \ No newline at end of file + + @Test + public void testWriteMessageQueueSkipsInvalidOrderTopicConfItems() { + topicRouteData.setOrderTopicConf("invalid;:2;" + BROKER_NAME + ":not-a-number;" + BROKER_NAME + ":0;" + + BROKER_NAME + ":-1;" + BROKER_NAME + ":2147483647;" + BROKER_NAME + ":2147483648;unknownBroker:1;" + + BROKER_NAME + ":2"); + + MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false); + + assertEquals(2, messageQueueSelector.getQueues().size()); + assertEquals(1, messageQueueSelector.getBrokerActingQueues().size()); + for (int i = 0; i < messageQueueSelector.getQueues().size(); i++) { + AddressableMessageQueue messageQueue = messageQueueSelector.getQueues().get(i); + assertEquals(BROKER_NAME, messageQueue.getBrokerName()); + assertEquals(i, messageQueue.getQueueId()); + } + } + + @Test + public void testGetMasterAddrReturnsEmptyForUnknownBroker() { + TopicRouteWrapper topicRouteWrapper = new TopicRouteWrapper(topicRouteData, TOPIC); + + assertNull(topicRouteWrapper.getMasterAddr("unknownBroker")); + assertFalse(topicRouteWrapper.getOptionalMasterAddr("unknownBroker").isPresent()); + } + + @Test + public void testWriteMessageQueueReturnsEmptyWhenAllOrderTopicConfItemsAreInvalid() { + topicRouteData.setOrderTopicConf("invalid;" + BROKER_NAME + ":0;" + BROKER_NAME + ":-1;" + + BROKER_NAME + ":2147483647;" + BROKER_NAME + ":2147483648;unknownBroker:1"); + + MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false); + + assertTrue(messageQueueSelector.getQueues().isEmpty()); + assertTrue(messageQueueSelector.getBrokerActingQueues().isEmpty()); + } + + @Test + public void testWriteMessageQueueSkipsQueueDataWithoutMasterAddr() { + queueData.setPerm(PermName.PERM_WRITE); + queueData.setWriteQueueNums(3); + queueData.setBrokerName("unknownBroker"); + + MessageQueueSelector messageQueueSelector = new MessageQueueSelector(new TopicRouteWrapper(topicRouteData, TOPIC), false); + + assertTrue(messageQueueSelector.getQueues().isEmpty()); + assertTrue(messageQueueSelector.getBrokerActingQueues().isEmpty()); + } +}