[ISSUE #8380]fix wrong election in controllerMode

controller/src/main/java/org/apache/rocketmq/controller/impl/manager/RaftReplicasInfoManager.java

@@ -123,12 +123,7 @@ public ControllerResult<CheckNotActiveBrokerResponse> checkNotActiveBroker(Check
                 log.warn("Broker expired, brokerInfo {}, expired {}ms", next.getKey(), timeoutMillis);
             }
         }
-        List<String> needReElectBrokerNames = scanNeedReelectBrokerSets(new BrokerValidPredicate() {
-            @Override
-            public boolean check(String clusterName, String brokerName, Long brokerId) {
-                return !isBrokerActive(clusterName, brokerName, brokerId, checkTime);
-            }
-        });
+        List<String> needReElectBrokerNames = scanNeedReelectBrokerSets((clusterName, brokerName, brokerId) -> isBrokerActive(clusterName, brokerName, brokerId, checkTime));
         Set<String> alreadyReportedBrokerName = notActiveBrokerIdentityInfoList.stream()
             .map(BrokerIdentityInfo::getBrokerName)
             .collect(Collectors.toSet());

controller/src/test/java/org/apache/rocketmq/controller/impl/manager/RaftReplicasInfoManagerTest.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.controller.impl.manager;
+
+import com.alibaba.fastjson.JSON;
+import org.apache.rocketmq.controller.impl.event.ControllerResult;
+import org.apache.rocketmq.controller.impl.heartbeat.BrokerIdentityInfo;
+import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
+import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerRequest;
+import org.apache.rocketmq.controller.impl.task.CheckNotActiveBrokerResponse;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_BROKER_NAME;
+import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_CLUSTER_NAME;
+import static org.apache.rocketmq.controller.ControllerTestBase.DEFAULT_IP;
+import static org.junit.Assert.assertEquals;
+
+public class RaftReplicasInfoManagerTest extends ReplicasInfoManagerTest {
+    private RaftReplicasInfoManager raftReplicasInfoManager;
+
+    @Before
+    public void init() {
+        super.init();
+        raftReplicasInfoManager = new RaftReplicasInfoManager(config);
+        replicasInfoManager = raftReplicasInfoManager;
+    }
+
+    @Test
+    public void testCheckNotActiveBroker() throws NoSuchFieldException, IllegalAccessException {
+        CheckNotActiveBrokerRequest request = new CheckNotActiveBrokerRequest();
+        long checkTimeMillis = request.getCheckTimeMillis() - 1000;
+        mockMetaData();
+        Field field = RaftReplicasInfoManager.class.getDeclaredField("brokerLiveTable");
+        field.setAccessible(true);
+        Map<BrokerIdentityInfo, BrokerLiveInfo> brokerLiveTable = (Map) field.get(raftReplicasInfoManager);
+        BrokerIdentityInfo broker1 = new BrokerIdentityInfo(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, 1L);
+        BrokerLiveInfo brokerLiveInfo1 = new BrokerLiveInfo(DEFAULT_BROKER_NAME, DEFAULT_IP[0], 1L, checkTimeMillis, 0L, null, 1, 0L, 1);
+        BrokerIdentityInfo broker2 = new BrokerIdentityInfo(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, 2L);
+        BrokerLiveInfo brokerLiveInfo2 = new BrokerLiveInfo(DEFAULT_BROKER_NAME, DEFAULT_IP[0], 1L, checkTimeMillis, 1000 * 60 * 20L, null, 1, 0L, 1);
+        BrokerIdentityInfo broker3 = new BrokerIdentityInfo(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, 3L);
+        BrokerLiveInfo brokerLiveInfo3 = new BrokerLiveInfo(DEFAULT_BROKER_NAME, DEFAULT_IP[0], 1L, checkTimeMillis, 1000 * 60 * 20L, null, 1, 0L, 1);
+        brokerLiveTable.put(broker1, brokerLiveInfo1);
+        brokerLiveTable.put(broker2, brokerLiveInfo2);
+        brokerLiveTable.put(broker3, brokerLiveInfo3);
+        ControllerResult<CheckNotActiveBrokerResponse> result = raftReplicasInfoManager.checkNotActiveBroker(request);
+        List<BrokerIdentityInfo> list2 = new ArrayList<>();
+        list2.add(broker1);
+        assertEquals(new String(result.getBody()), new String(JSON.toJSONBytes(list2)));
+    }
+}

controller/src/test/java/org/apache/rocketmq/controller/impl/manager/ReplicasInfoManagerTest.java

@@ -64,11 +64,11 @@
 import static org.junit.Assert.assertTrue;
 
 public class ReplicasInfoManagerTest {
-    private ReplicasInfoManager replicasInfoManager;
+    protected ReplicasInfoManager replicasInfoManager;
 
-    private DefaultBrokerHeartbeatManager heartbeatManager;
+    protected DefaultBrokerHeartbeatManager heartbeatManager;
 
-    private ControllerConfig config;
+    protected ControllerConfig config;
 
     @Before
     public void init() {
@@ -508,4 +508,18 @@ public void testSerialize() {
             Field[] fields = oldReplicaInfo.getClass().getFields();
         }
     }
+
+    @Test
+    public void testScanNeedReelectBrokerSets() {
+        mockMetaData();
+        List<String> strings = this.replicasInfoManager.scanNeedReelectBrokerSets(
+                (clusterName, brokerName, brokerId) -> mockValidPredicate1(brokerId));
+        assertArrayEquals(strings.toArray(),new String[]{DEFAULT_BROKER_NAME});
+    }
+
+    private boolean mockValidPredicate1(Long brokerId) {
+        // mock only broker1(id=2) & broker2(id=3) is alive,master(id=1) is inactive
+        List<Long> inactiveBroker = Arrays.asList(2L,3L);
+        return inactiveBroker.contains(brokerId);
+    }
 }