public ControllerResult<Void> cleanBrokerData(final CleanControllerBrokerDataRequestHeader requestHeader,
        final BrokerValidPredicate validPredicate) {
        final ControllerResult<Void> result = new ControllerResult<>();

        final String clusterName = requestHeader.getClusterName();
        final String brokerName = requestHeader.getBrokerName();
        final String brokerControllerIdsToClean = requestHeader.getBrokerControllerIdsToClean();

        Set<Long> brokerIdSet = null;
        if (!requestHeader.isCleanLivingBroker()) {
            //if SyncStateInfo.masterAddress is not empty, at least one broker with the same BrokerName is alive
            SyncStateInfo syncStateInfo = this.syncStateSetInfoTable.get(brokerName);
            if (StringUtils.isBlank(brokerControllerIdsToClean) && null != syncStateInfo && syncStateInfo.getMasterBrokerId() != null) {
                String remark = String.format("Broker %s is still alive, clean up failure", requestHeader.getBrokerName());
                result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, remark);
                return result;
            }
            if (StringUtils.isNotBlank(brokerControllerIdsToClean)) {
                try {
                    brokerIdSet = Stream.of(brokerControllerIdsToClean.split(";")).map(idStr -> Long.valueOf(idStr)).collect(Collectors.toSet());
                } catch (NumberFormatException numberFormatException) {
                    String remark = String.format("Please set the option <brokerControllerIdsToClean> according to the format, exception: %s", numberFormatException);
                    result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, remark);
                    return result;
                }
                for (Long brokerId : brokerIdSet) {
                    if (validPredicate.check(clusterName, brokerName, brokerId)) {
                        String remark = String.format("Broker [%s,  %s] is still alive, clean up failure", requestHeader.getBrokerName(), brokerId);
                        result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, remark);
                        return result;
                    }
                }
            }
        }
        if (isContainsBroker(brokerName)) {
            final CleanBrokerDataEvent event = new CleanBrokerDataEvent(brokerName, brokerIdSet);
            result.addEvent(event);
            return result;
        }
        result.setCodeAndRemark(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, String.format("Broker %s is not existed,clean broker data failure.", brokerName));
        return result;
    }
@Test
    public void testCleanBrokerData() {
        mockMetaData();
        CleanControllerBrokerDataRequestHeader header1 = new CleanControllerBrokerDataRequestHeader(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, "1");
        ControllerResult<Void> result1 = this.replicasInfoManager.cleanBrokerData(header1, (cluster, brokerName, brokerId) -> true);
        assertEquals(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, result1.getResponseCode());

        CleanControllerBrokerDataRequestHeader header2 = new CleanControllerBrokerDataRequestHeader(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, null);
        ControllerResult<Void> result2 = this.replicasInfoManager.cleanBrokerData(header2, (cluster, brokerName, brokerId) -> true);
        assertEquals(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, result2.getResponseCode());
        assertEquals("Broker broker-set-a is still alive, clean up failure", result2.getRemark());

        CleanControllerBrokerDataRequestHeader header3 = new CleanControllerBrokerDataRequestHeader(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, "1");
        ControllerResult<Void> result3 = this.replicasInfoManager.cleanBrokerData(header3, (cluster, brokerName, brokerId) -> false);
        assertEquals(ResponseCode.SUCCESS, result3.getResponseCode());

        CleanControllerBrokerDataRequestHeader header4 = new CleanControllerBrokerDataRequestHeader(DEFAULT_CLUSTER_NAME, DEFAULT_BROKER_NAME, "1;2;3");
        ControllerResult<Void> result4 = this.replicasInfoManager.cleanBrokerData(header4, (cluster, brokerName, brokerId) -> false);
        assertEquals(ResponseCode.SUCCESS, result4.getResponseCode());

        CleanControllerBrokerDataRequestHeader header5 = new CleanControllerBrokerDataRequestHeader(DEFAULT_CLUSTER_NAME, "broker12", "1;2;3", true);
        ControllerResult<Void> result5 = this.replicasInfoManager.cleanBrokerData(header5, (cluster, brokerName, brokerId) -> false);
        assertEquals(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, result5.getResponseCode());
        assertEquals("Broker broker12 is not existed,clean broker data failure.", result5.getRemark());

        CleanControllerBrokerDataRequestHeader header6 = new CleanControllerBrokerDataRequestHeader(null, "broker12", "1;2;3", true);
        ControllerResult<Void> result6 = this.replicasInfoManager.cleanBrokerData(header6, (cluster, brokerName, brokerId) -> cluster != null);
        assertEquals(ResponseCode.CONTROLLER_INVALID_CLEAN_BROKER_METADATA, result6.getResponseCode());

        CleanControllerBrokerDataRequestHeader header7 = new CleanControllerBrokerDataRequestHeader(null, DEFAULT_BROKER_NAME, "1;2;3", true);
        ControllerResult<Void> result7 = this.replicasInfoManager.cleanBrokerData(header7, (cluster, brokerName, brokerId) -> false);
        assertEquals(ResponseCode.SUCCESS, result7.getResponseCode());

    }