[ISSUE #8974] Add feature switch of recalling, disable by default (#9067

)
apache · Dec 27, 2024 · 19393e0 · 19393e0
1 parent 2089abd
commit 19393e0
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 0 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/RecallMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/RecallMessageProcessor.java
@@ -57,6 +57,12 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand
         final RecallMessageRequestHeader requestHeader =
             request.decodeCommandCustomHeader(RecallMessageRequestHeader.class);
 
+        if (!brokerController.getBrokerConfig().isRecallMessageEnable()) {
+            response.setCode(ResponseCode.NO_PERMISSION);
+            response.setRemark("recall failed, operation is forbidden");
+            return response;
+        }
+
         if (BrokerRole.SLAVE == brokerController.getMessageStoreConfig().getBrokerRole()) {
             response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
             response.setRemark("recall failed, broker service not available");

diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/RecallMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/RecallMessageProcessorTest.java
@@ -89,6 +89,7 @@ public void init() throws IllegalAccessException, NoSuchFieldException {
         when(brokerController.getMessageStore()).thenReturn(messageStore);
         when(brokerController.getBrokerConfig()).thenReturn(brokerConfig);
         when(brokerConfig.getBrokerName()).thenReturn(BROKER_NAME);
+        when(brokerConfig.isRecallMessageEnable()).thenReturn(true);
         when(brokerController.getBrokerStatsManager()).thenReturn(brokerStatsManager);
         when(handlerContext.channel()).thenReturn(channel);
         recallMessageProcessor = new RecallMessageProcessor(brokerController);
@@ -134,6 +135,14 @@ public void testHandlePutMessageResult() {
         }
     }
 
+    @Test
+    public void testProcessRequest_notEnable() throws RemotingCommandException {
+        when(brokerConfig.isRecallMessageEnable()).thenReturn(false);
+        RemotingCommand request = mockRequest(0, TOPIC, TOPIC, "id", BROKER_NAME);
+        RemotingCommand response = recallMessageProcessor.processRequest(handlerContext, request);
+        Assert.assertEquals(ResponseCode.NO_PERMISSION, response.getCode());
+    }
+
     @Test
     public void testProcessRequest_invalidStatus() throws RemotingCommandException {
         RemotingCommand request = mockRequest(0, TOPIC, TOPIC, "id", BROKER_NAME);

diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -453,6 +453,8 @@ public class BrokerConfig extends BrokerIdentity {
 
     private boolean allowRecallWhenBrokerNotWriteable = true;
 
+    private boolean recallMessageEnable = false;
+
     public String getConfigBlackList() {
         return configBlackList;
     }
@@ -1996,4 +1998,12 @@ public boolean isAllowRecallWhenBrokerNotWriteable() {
     public void setAllowRecallWhenBrokerNotWriteable(boolean allowRecallWhenBrokerNotWriteable) {
         this.allowRecallWhenBrokerNotWriteable = allowRecallWhenBrokerNotWriteable;
     }
+
+    public boolean isRecallMessageEnable() {
+        return recallMessageEnable;
+    }
+
+    public void setRecallMessageEnable(boolean recallMessageEnable) {
+        this.recallMessageEnable = recallMessageEnable;
+    }
 }
diff --git a/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java b/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
@@ -138,6 +138,7 @@ public static BrokerController createAndStartBroker(String nsAddr) {
         brokerConfig.setEnableCalcFilterBitMap(true);
         brokerConfig.setAppendAckAsync(true);
         brokerConfig.setAppendCkAsync(true);
+        brokerConfig.setRecallMessageEnable(true);
         storeConfig.setEnableConsumeQueueExt(true);
         brokerConfig.setLoadBalancePollNameServerInterval(500);
         storeConfig.setStorePathRootDir(baseDir);