fix(broker): fix wrong test

1. fix wrong test

broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java

@@ -84,7 +84,7 @@ public class ReplicasManager {
     private volatile String controllerLeaderAddress = "";
     private volatile State state = State.INITIAL;
 
-    private RegisterState registerState = RegisterState.INITIAL;
+    private volatile RegisterState registerState = RegisterState.INITIAL;
 
     private ScheduledFuture<?> checkSyncStateSetTaskFuture;
     private ScheduledFuture<?> slaveSyncFuture;
@@ -191,6 +191,7 @@ public class ReplicasManager {
             }
             // register 5 times but still unsuccessful
             if (this.state != State.REGISTER_TO_CONTROLLER_DONE) {
+                LOGGER.error("Register to broker failed 5 times");
                 return false;
             }
         }

controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java

@@ -21,6 +21,11 @@ import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
 
 public interface BrokerHeartbeatManager {
 
+    /**
+     * initialize the resources
+     * @return
+     */
+    void initialize();
     /**
      * Broker new heartbeat.
      */

controller/src/main/java/org/apache/rocketmq/controller/ControllerManager.java

@@ -76,6 +76,7 @@ public class ControllerManager {
         this.configuration = new Configuration(log, this.controllerConfig, this.nettyServerConfig);
         this.configuration.setStorePathFromConfig(this.controllerConfig, "configStorePath");
         this.remotingClient = new NettyRemotingClient(nettyClientConfig);
+        this.heartbeatManager = new DefaultBrokerHeartbeatManager(this.controllerConfig);
     }
 
     public boolean initialize() {
@@ -92,7 +93,6 @@ public class ControllerManager {
                 return new FutureTaskExt<T>(runnable, value);
             }
         };
-        this.heartbeatManager = new DefaultBrokerHeartbeatManager(this.controllerConfig);
         if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerPeers())) {
             throw new IllegalArgumentException("Attribute value controllerDLegerPeers of ControllerConfig is null or empty");
         }
@@ -103,6 +103,9 @@ public class ControllerManager {
             this.nettyServerConfig, this.nettyClientConfig, this.brokerHousekeepingService,
             new DefaultElectPolicy(this.heartbeatManager::isBrokerActive, this.heartbeatManager::getBrokerLiveInfo));
 
+        // Initialize the basic resources
+        this.heartbeatManager.initialize();
+
         // Register broker inactive listener
         this.heartbeatManager.addBrokerLifecycleListener(this::onBrokerInactive);
         registerProcessor();

controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java

@@ -38,8 +38,8 @@ import org.apache.rocketmq.remoting.common.RemotingHelper;
 public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.CONTROLLER_LOGGER_NAME);
     private static final long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 10;
-    private final ScheduledExecutorService scheduledService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("DefaultBrokerHeartbeatManager_scheduledService_"));
-    private final ExecutorService executor = Executors.newFixedThreadPool(2, new ThreadFactoryImpl("DefaultBrokerHeartbeatManager_executorService_"));
+    private ScheduledExecutorService scheduledService;
+    private ExecutorService executor;
 
     private final ControllerConfig controllerConfig;
     private final Map<BrokerIdentityInfo/* brokerIdentity*/, BrokerLiveInfo> brokerLiveTable;
@@ -62,6 +62,12 @@ public class DefaultBrokerHeartbeatManager implements BrokerHeartbeatManager {
         this.executor.shutdown();
     }
 
+    @Override
+    public void initialize() {
+        this.scheduledService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("DefaultBrokerHeartbeatManager_scheduledService_"));
+        this.executor = Executors.newFixedThreadPool(2, new ThreadFactoryImpl("DefaultBrokerHeartbeatManager_executorService_"));
+    }
+
     public void scanNotActiveBroker() {
         try {
             log.info("start scanNotActiveBroker");

test/src/test/java/org/apache/rocketmq/test/autoswitchrole/AutoSwitchRoleIntegrationTest.java

@@ -60,6 +60,9 @@ public class AutoSwitchRoleIntegrationTest extends AutoSwitchRoleBase {
     private static ControllerManager controllerManager;
     private static String nameserverAddress;
     private static String controllerAddress;
+
+    private static ControllerConfig controllerConfig;
+
     private BrokerController brokerController1;
     private BrokerController brokerController2;
     private Random random = new Random();
@@ -75,19 +78,23 @@ public class AutoSwitchRoleIntegrationTest extends AutoSwitchRoleBase {
         int namesrvPort = nextPort();
         serverConfig.setListenPort(namesrvPort);
 
-        ControllerConfig controllerConfig = buildControllerConfig("n0", peers);
+        controllerConfig = buildControllerConfig("n0", peers);
         namesrvController = new NamesrvController(new NamesrvConfig(), serverConfig, new NettyClientConfig());
         assertTrue(namesrvController.initialize());
         namesrvController.start();
 
-        controllerManager = new ControllerManager(controllerConfig, new NettyServerConfig(), new NettyClientConfig());
-        assertTrue(controllerManager.initialize());
-        controllerManager.start();
+        initAndStartControllerManager();
 
         nameserverAddress = "127.0.0.1:" + namesrvPort + ";";
         controllerAddress = "127.0.0.1:" + controllerPort + ";";
     }
 
+    private static void initAndStartControllerManager() {
+        controllerManager = new ControllerManager(controllerConfig, new NettyServerConfig(), new NettyClientConfig());
+        assertTrue(controllerManager.initialize());
+        controllerManager.start();
+    }
+
     public void initBroker(int mappedFileSize, String brokerName) throws Exception {
 
         this.brokerController1 = startBroker(nameserverAddress, controllerAddress, brokerName, 1, nextPort(), nextPort(), nextPort(), BrokerRole.SYNC_MASTER, mappedFileSize);
@@ -148,6 +155,8 @@ public class AutoSwitchRoleIntegrationTest extends AutoSwitchRoleBase {
         String topic = "Topic-" + AutoSwitchRoleIntegrationTest.class.getSimpleName() + random.nextInt(65535);
         String brokerName = "Broker-" + AutoSwitchRoleIntegrationTest.class.getSimpleName() + random.nextInt(65535);
         initBroker(DEFAULT_FILE_SIZE, brokerName);
+        int listenPort = brokerController1.getBrokerConfig().getListenPort();
+        int nettyPort = brokerController1.getNettyServerConfig().getListenPort();
         mockData(topic);
 
         // Let master shutdown
@@ -160,7 +169,7 @@ public class AutoSwitchRoleIntegrationTest extends AutoSwitchRoleBase {
         assertEquals(brokerController2.getReplicasManager().getMasterEpoch(), 2);
 
         // Restart old master, it should be slave
-        brokerController1 = startBroker(nameserverAddress, controllerAddress, brokerName, 1, nextPort(), nextPort(), nextPort(), BrokerRole.SLAVE, DEFAULT_FILE_SIZE);
+        brokerController1 = startBroker(nameserverAddress, controllerAddress, brokerName, 1, nextPort(), listenPort, nettyPort, BrokerRole.SLAVE, DEFAULT_FILE_SIZE);
         waitSlaveReady(brokerController1.getMessageStore());
 
         assertFalse(brokerController1.getReplicasManager().isMasterState());
@@ -224,6 +233,8 @@ public class AutoSwitchRoleIntegrationTest extends AutoSwitchRoleBase {
         // Put message from 10 to 19
         putMessage(this.brokerController1.getMessageStore(), topic);
         checkMessage(this.brokerController2.getMessageStore(), topic, 20, 0);
+
+        initAndStartControllerManager();
     }
 
     @Test