public ControllerResult<ElectMasterResponseHeader> electMaster(final ElectMasterRequestHeader request,
        final ElectPolicy electPolicy) {
        final String brokerName = request.getBrokerName();
        final Long brokerId = request.getBrokerId();
        final ControllerResult<ElectMasterResponseHeader> result = new ControllerResult<>(new ElectMasterResponseHeader());
        final ElectMasterResponseHeader response = result.getResponse();
        if (!isContainsBroker(brokerName)) {
            // this broker set hasn't been registered
            result.setCodeAndRemark(ResponseCode.CONTROLLER_BROKER_NEED_TO_BE_REGISTERED, "Broker hasn't been registered");
            return result;
        }

        final SyncStateInfo syncStateInfo = this.syncStateSetInfoTable.get(brokerName);
        final BrokerReplicaInfo brokerReplicaInfo = this.replicaInfoTable.get(brokerName);
        final Set<Long> syncStateSet = syncStateInfo.getSyncStateSet();
        final Long oldMaster = syncStateInfo.getMasterBrokerId();
        Set<Long> allReplicaBrokers = controllerConfig.isEnableElectUncleanMaster() ? brokerReplicaInfo.getAllBroker() : null;
        Long newMaster = null;

        if (syncStateInfo.isFirstTimeForElect()) {
            // If never have a master in this broker set, in other words, it is the first time to elect a master
            // elect it as the first master
            newMaster = brokerId;
        }

        // elect by policy
        if (newMaster == null || newMaster == -1) {
            // we should assign this assignedBrokerId when the brokerAddress need to be elected by force
            Long assignedBrokerId = request.getDesignateElect() ? brokerId : null;
            newMaster = electPolicy.elect(brokerReplicaInfo.getClusterName(), brokerReplicaInfo.getBrokerName(), syncStateSet, allReplicaBrokers, oldMaster, assignedBrokerId);
        }

        if (newMaster != null && newMaster.equals(oldMaster)) {
            // old master still valid, change nothing
            String err = String.format("The old master %s is still alive, not need to elect new master for broker %s", oldMaster, brokerReplicaInfo.getBrokerName());
            LOGGER.warn("{}", err);
            // the master still exist
            response.setMasterEpoch(syncStateInfo.getMasterEpoch());
            response.setSyncStateSetEpoch(syncStateInfo.getSyncStateSetEpoch());
            response.setMasterBrokerId(oldMaster);
            response.setMasterAddress(brokerReplicaInfo.getBrokerAddress(oldMaster));

            result.setBody(new ElectMasterResponseBody(syncStateSet).encode());
            result.setCodeAndRemark(ResponseCode.CONTROLLER_MASTER_STILL_EXIST, err);
            return result;
        }

        // a new master is elected
        if (newMaster != null) {
            final int masterEpoch = syncStateInfo.getMasterEpoch();
            final int syncStateSetEpoch = syncStateInfo.getSyncStateSetEpoch();
            final HashSet<Long> newSyncStateSet = new HashSet<>();
            newSyncStateSet.add(newMaster);

            response.setMasterBrokerId(newMaster);
            response.setMasterAddress(brokerReplicaInfo.getBrokerAddress(newMaster));
            response.setMasterEpoch(masterEpoch + 1);
            response.setSyncStateSetEpoch(syncStateSetEpoch + 1);
            ElectMasterResponseBody responseBody = new ElectMasterResponseBody(newSyncStateSet);

            BrokerMemberGroup brokerMemberGroup = buildBrokerMemberGroup(brokerReplicaInfo);
            if (null != brokerMemberGroup) {
                responseBody.setBrokerMemberGroup(brokerMemberGroup);
            }

            result.setBody(responseBody.encode());
            final ElectMasterEvent event = new ElectMasterEvent(brokerName, newMaster);
            result.addEvent(event);
            LOGGER.info("Elect new master {} for broker {}", newMaster, brokerName);
            return result;
        }
        // If elect failed and the electMaster is triggered by controller (we can figure it out by brokerAddress),
        // we still need to apply an ElectMasterEvent to tell the statemachine
        // that the master was shutdown and no new master was elected.
        if (request.getBrokerId() == null || request.getBrokerId() == -1) {
            final ElectMasterEvent event = new ElectMasterEvent(false, brokerName);
            result.addEvent(event);
            result.setCodeAndRemark(ResponseCode.CONTROLLER_MASTER_NOT_AVAILABLE, "Old master has down and failed to elect a new broker master");
        } else {
            result.setCodeAndRemark(ResponseCode.CONTROLLER_ELECT_MASTER_FAILED, "Failed to elect a new master");
        }
        LOGGER.warn("Failed to elect a new master for broker {}", brokerName);
        return result;
    }
@Test
    public void testElectMasterPreferHigherOffsetWhenEpochEquals() {
        mockMetaData();
        final ElectMasterRequestHeader request = ElectMasterRequestHeader.ofControllerTrigger(DEFAULT_BROKER_NAME);
        ElectPolicy electPolicy = new DefaultElectPolicy(this.heartbeatManager::isBrokerActive, this.heartbeatManager::getBrokerLiveInfo);
        mockHeartbeatDataHigherOffset();
        final ControllerResult<ElectMasterResponseHeader> cResult = this.replicasInfoManager.electMaster(request,
            electPolicy);
        final ElectMasterResponseHeader response = cResult.getResponse();
        assertEquals(DEFAULT_IP[2], response.getMasterAddress());
        assertEquals(3L, response.getMasterBrokerId().longValue());
        assertEquals(2, response.getMasterEpoch().intValue());
    }