add timeout & log

Author: xiaoming.lt

broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java

@@ -778,20 +778,28 @@ public TopicConfigAndMappingSerializeWrapper getAllTopicConfig(final String addr
 
         DataVersion topicConfigDataVersion = null;
         DataVersion mappingDataVersion = null;
+        long timeoutMills = getTimeoutMillis();
         int topicSeq = 0;
+        long beginTime = System.nanoTime();
         ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
         Map<String, TopicQueueMappingDetail> topicQueueMappingDetailMap = new ConcurrentHashMap<>();
         while (true) {
+            long leftTime = timeoutMills - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime);
+            if (leftTime < 0) {
+                throw new RemotingTimeoutException("invokeSync call timeout");
+            }
+
             GetAllTopicConfigRequestHeader requestHeader = new GetAllTopicConfigRequestHeader();
             requestHeader.setTopicSeq(topicSeq);
-            requestHeader.setMaxTopicNum(2000);
+            requestHeader.setMaxTopicNum(getMaxPageSize());
             requestHeader.setDataVersion(Optional.ofNullable(topicConfigDataVersion).
                 map(DataVersion::toJson).orElse(StringUtils.EMPTY));
-
+            LOGGER.info("getAllTopicConfig from seq {}, max {}, dataVersion {}",
+                    topicSeq, requestHeader.getMaxTopicNum(), requestHeader.getDataVersion());
             RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, requestHeader);
 
             RemotingCommand response = this.remotingClient.invokeSync(
-                MixAll.brokerVIPChannel(true, addr), request, 3000);
+                MixAll.brokerVIPChannel(true, addr), request, 30000);
 
             assert response != null;
             if (response.getCode() == SUCCESS) {
@@ -919,20 +927,29 @@ public String getAllDelayOffset(
     public SubscriptionGroupWrapper getAllSubscriptionGroupConfig(final String addr)
         throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
         RemotingConnectException, MQBrokerException, RemotingCommandException {
+
+        long timeoutMills = getTimeoutMillis();
         DataVersion currentDataVersion = null;
         int groupSeq = 0;
+        long beginTime = System.nanoTime();
         ConcurrentMap<String, SubscriptionGroupConfig> subscriptionGroupTable = new ConcurrentHashMap<>();
         ConcurrentMap<String, ConcurrentMap<String, Integer>> forbiddenTable = new ConcurrentHashMap<>();
+
         while (true) {
+            long leftTime = timeoutMills - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime);
+            if (leftTime < 0) {
+                throw new RemotingTimeoutException("invokeSync call timeout");
+            }
 
             GetAllSubscriptionGroupRequestHeader requestHeader = new GetAllSubscriptionGroupRequestHeader();
             requestHeader.setGroupSeq(groupSeq);
-            requestHeader.setMaxGroupNum(2000);
+            requestHeader.setMaxGroupNum(getMaxPageSize());
             requestHeader.setDataVersion(Optional.ofNullable(currentDataVersion)
                 .map(DataVersion::toJson).orElse(StringUtils.EMPTY));
-
+            LOGGER.info("getAllSubscriptionGroup from seq {}, max {}, dataVersion {}",
+                    groupSeq, requestHeader.getMaxGroupNum(), requestHeader.getDataVersion());
             RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, requestHeader);
-            RemotingCommand response = this.remotingClient.invokeSync(addr, request, 3000);
+            RemotingCommand response = this.remotingClient.invokeSync(addr, request, 30000);
 
             assert response != null;
             if (response.getCode() == SUCCESS) {
@@ -1610,4 +1627,12 @@ private PullResult processPullResult(final PullResultExt pullResult, String brok
         return pullResult;
     }
 
+    private int getMaxPageSize() {
+        return 2000;
+    }
+
+    private long getTimeoutMillis() {
+        return TimeUnit.SECONDS.toMillis(60);
+    }
+
 }

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -2836,7 +2836,8 @@ public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr,
             requestHeader.setMaxGroupNum(clientConfig.getMaxPageSizeInGetMetadata());
             requestHeader.setDataVersion(Optional.ofNullable(currentDataVersion)
                 .map(DataVersion::toJson).orElse(StringUtils.EMPTY));
-
+            log.info("getAllSubscriptionGroup from seq {}, max {}, dataVersion {}",
+                    groupSeq, requestHeader.getMaxGroupNum(), requestHeader.getDataVersion());
             RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, requestHeader);
             RemotingCommand response = this.remotingClient.invokeSync(
                 MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), brokerAddr), request, leftTime);
@@ -2930,7 +2931,8 @@ public TopicConfigSerializeWrapper getAllTopicConfig(final String addr, long tim
             requestHeader.setMaxTopicNum(clientConfig.getMaxPageSizeInGetMetadata());
             requestHeader.setDataVersion(Optional.ofNullable(currentDataVersion).
                 map(DataVersion::toJson).orElse(StringUtils.EMPTY));
-
+            log.info("getAllTopicConfig from seq {}, max {}, dataVersion {}",
+                    topicSeq, requestHeader.getMaxTopicNum(), requestHeader.getDataVersion());
             RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, requestHeader);
 
             RemotingCommand response = this.remotingClient.invokeSync(