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

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

@@ -57,6 +57,12 @@ public class RecallMessageProcessor implements NettyRequestProcessor {
         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");

broker/src/test/java/org/apache/rocketmq/broker/processor/RecallMessageProcessorTest.java

@@ -89,6 +89,7 @@ public class RecallMessageProcessorTest {
         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 class RecallMessageProcessorTest {
         }
     }
 
+    @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);

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 class BrokerConfig extends BrokerIdentity {
     public void setAllowRecallWhenBrokerNotWriteable(boolean allowRecallWhenBrokerNotWriteable) {
         this.allowRecallWhenBrokerNotWriteable = allowRecallWhenBrokerNotWriteable;
     }
+
+    public boolean isRecallMessageEnable() {
+        return recallMessageEnable;
+    }
+
+    public void setRecallMessageEnable(boolean recallMessageEnable) {
+        this.recallMessageEnable = recallMessageEnable;
+    }
 }

test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java

@@ -138,6 +138,7 @@ public class IntegrationTestBase {
         brokerConfig.setEnableCalcFilterBitMap(true);
         brokerConfig.setAppendAckAsync(true);
         brokerConfig.setAppendCkAsync(true);
+        brokerConfig.setRecallMessageEnable(true);
         storeConfig.setEnableConsumeQueueExt(true);
         brokerConfig.setLoadBalancePollNameServerInterval(500);
         storeConfig.setStorePathRootDir(baseDir);