[ISSUE apache#8657] Make retry topic pop probability configurable

broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java

@@ -373,7 +373,7 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingC
         // considered the same type because they share the same retry flag in previous fields.
         // Therefore, needRetryV1 is designed as a subset of needRetry, and within a single request,
         // only one type of retry topic is able to call popMsgFromQueue.
-        boolean needRetry = randomQ % 5 == 0;
+        boolean needRetry = randomQ < brokerConfig.getPopFromRetryProbability();
         boolean needRetryV1 = false;
         if (brokerConfig.isEnableRetryTopicV2() && brokerConfig.isRetrieveMessageFromPopRetryTopicV1()) {
             needRetryV1 = randomQ % 2 == 0;

common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java

@@ -231,7 +231,7 @@ public class BrokerConfig extends BrokerIdentity {
     // read message from pop retry topic v1, for the compatibility, will be removed in the future version
     private boolean retrieveMessageFromPopRetryTopicV1 = true;
     private boolean enableRetryTopicV2 = false;
-
+    private int popFromRetryProbability = 20;
     private boolean realTimeNotifyConsumerChange = true;
 
     private boolean litePullMessageEnable = true;
@@ -563,6 +563,15 @@ public void setEnablePopLog(boolean enablePopLog) {
         this.enablePopLog = enablePopLog;
     }
 
+    public int getPopFromRetryProbability() {
+        return popFromRetryProbability;
+    }
+
+    public void setPopFromRetryProbability(int popFromRetryProbability) {
+        this.popFromRetryProbability = popFromRetryProbability;
+    }
+
+
     public boolean isTraceOn() {
         return traceOn;
     }