[ISSUE #10688] Skip malformed ordered route items

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<Address>
     @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

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<AddressableMessageQueue> queues = new ArrayList<>();
@@ -110,17 +116,27 @@ private static List<AddressableMessageQueue> 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<String> 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<Integer> 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<AddressableMessageQueue> buildWrite(TopicRouteWrapper topicR
 
             for (QueueData qd : qds) {
                 if (PermName.isWriteable(qd.getPerm())) {
-                    String brokerAddr = topicRoute.getMasterAddr(qd.getBrokerName());
-                    if (brokerAddr == null) {
+                    Optional<String> 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<AddressableMessageQueue> buildWrite(TopicRouteWrapper topicR
         return queueSet.stream().sorted().collect(Collectors.toList());
     }
 
+    private static Optional<Integer> 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<AddressableMessageQueue> normalQueues) {
         for (AddressableMessageQueue mq : normalQueues) {
             AddressableMessageQueue brokerActingQueue = new AddressableMessageQueue(

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<String> 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) {

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());
+        }
+    }
 }

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();

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));
     }
-}
+
+    @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());
+    }
+}