[ISSUE #7029] Add a config to determine whether pop response should r…

…eturn the actual retry topic or tamper with the original topic (#7030)
apache · Jul 18, 2023 · 737c1e5 · 737c1e5
1 parent 440be1e
commit 737c1e5
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 2 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
@@ -591,8 +591,8 @@ private CompletableFuture<Long> popMsgFromQueue(String attemptId, boolean isRetr
                 atomicRestNum.set(result.getMaxOffset() - result.getNextBeginOffset() + atomicRestNum.get());
                 String brokerName = brokerController.getBrokerConfig().getBrokerName();
                 for (SelectMappedBufferResult mapedBuffer : result.getMessageMapedList()) {
-                    // We should not recode buffer for normal topic message
-                    if (!isRetry) {
+                    // We should not recode buffer when popResponseReturnActualRetryTopic is true or topic is not retry topic
+                    if (brokerController.getBrokerConfig().isPopResponseReturnActualRetryTopic() || !isRetry) {
                         getMessageResult.addMessage(mapedBuffer);
                     } else {
                         List<MessageExt> messageExtList = MessageDecoder.decodesBatch(mapedBuffer.getByteBuffer(),

diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -381,6 +381,11 @@ public class BrokerConfig extends BrokerIdentity {
      */
     private long fetchNamesrvAddrInterval = 10 * 1000;
 
+    /**
+     * Pop response returns the actual retry topic rather than tampering with the original topic
+     */
+    private boolean popResponseReturnActualRetryTopic = false;
+
     public long getMaxPopPollingSize() {
         return maxPopPollingSize;
     }
@@ -1676,4 +1681,12 @@ public long getFetchNamesrvAddrInterval() {
     public void setFetchNamesrvAddrInterval(final long fetchNamesrvAddrInterval) {
         this.fetchNamesrvAddrInterval = fetchNamesrvAddrInterval;
     }
+
+    public boolean isPopResponseReturnActualRetryTopic() {
+        return popResponseReturnActualRetryTopic;
+    }
+
+    public void setPopResponseReturnActualRetryTopic(boolean popResponseReturnActualRetryTopic) {
+        this.popResponseReturnActualRetryTopic = popResponseReturnActualRetryTopic;
+    }
 }