@Override
    public List<Long> getAllAvailableNodes() {
        List<Long> nodeIds = Lists.newArrayList();
        if (usedComputeNode || availableID2Backend.isEmpty()) {
            nodeIds.addAll(availableID2ComputeNode.keySet());
        }

        if (preferComputeNode) {
            return nodeIds;
        }
        nodeIds.addAll(availableID2Backend.keySet());
        return nodeIds;
    }
@Test
    public void testSelectBackendAndComputeNode() {
        new MockUp<SystemInfoService>() {
            @Mock
            public ImmutableMap<Long, ComputeNode> getIdToBackend() {
                return availableId2Backend;
            }

            @Mock
            public ImmutableMap<Long, ComputeNode> getIdComputeNode() {
                return availableId2ComputeNode;
            }
        };

        DefaultWorkerProvider.Factory workerProviderFactory = new DefaultWorkerProvider.Factory();
        DefaultWorkerProvider workerProvider;
        List<Integer> numUsedComputeNodesList = ImmutableList.of(-1, 0, 2, 3, 5, 8, 10);

        // test ComputeNode only
        for (Integer numUsedComputeNodes : numUsedComputeNodesList) {
            workerProvider =
                    workerProviderFactory.captureAvailableWorkers(GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo(),
                            true, numUsedComputeNodes, ComputationFragmentSchedulingPolicy.COMPUTE_NODES_ONLY, 
                            WarehouseManager.DEFAULT_WAREHOUSE_ID);
            List<Long> selectedWorkerIdsList = workerProvider.getAllAvailableNodes();
            for (Long selectedWorkerId : selectedWorkerIdsList) {
                Assert.assertTrue("selectedWorkerId:" + selectedWorkerId,
                        availableId2ComputeNode.containsKey(selectedWorkerId));
            }
        }
        // test Backend only
        for (Integer numUsedComputeNodes : numUsedComputeNodesList) {
            workerProvider =
                    workerProviderFactory.captureAvailableWorkers(GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo(),
                            false, numUsedComputeNodes, ComputationFragmentSchedulingPolicy.COMPUTE_NODES_ONLY, 
                            WarehouseManager.DEFAULT_WAREHOUSE_ID);
            List<Long> selectedWorkerIdsList = workerProvider.getAllAvailableNodes();
            Assert.assertEquals(availableId2Backend.size(), selectedWorkerIdsList.size());
            for (Long selectedWorkerId : selectedWorkerIdsList) {
                Assert.assertTrue("selectedWorkerId:" + selectedWorkerId,
                        availableId2Backend.containsKey(selectedWorkerId));
            }
        }
        // test Backend and ComputeNode
        for (Integer numUsedComputeNodes : numUsedComputeNodesList) {
            workerProvider =
                    workerProviderFactory.captureAvailableWorkers(GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo(),
                            true, numUsedComputeNodes, ComputationFragmentSchedulingPolicy.ALL_NODES, 
                            WarehouseManager.DEFAULT_WAREHOUSE_ID);
            List<Long> selectedWorkerIdsList = workerProvider.getAllAvailableNodes();
            Collections.reverse(selectedWorkerIdsList); //put ComputeNode id to the front,Backend id to the back
            //test ComputeNode
            for (int i = 0; i < availableId2ComputeNode.size() && i < selectedWorkerIdsList.size(); i++) {
                Assert.assertTrue("selectedWorkerId:" + selectedWorkerIdsList.get(i),
                        availableId2ComputeNode.containsKey(selectedWorkerIdsList.get(i)));
            }
            //test Backend
            for (int i = availableId2ComputeNode.size(); i < selectedWorkerIdsList.size(); i++) {
                Assert.assertTrue("selectedWorkerId:" + selectedWorkerIdsList.get(i),
                        availableId2Backend.containsKey(selectedWorkerIdsList.get(i)));
            }
        }
    }