@Override
    public ComputeNode getWorkerById(long workerId) {
        ComputeNode worker = availableID2Backend.get(workerId);
        if (worker != null) {
            return worker;
        }
        return availableID2ComputeNode.get(workerId);
    }
@Test
    public void testCaptureAvailableWorkers() {

        long deadBEId = 1L;
        long deadCNId = 11L;
        long inBlacklistBEId = 3L;
        long inBlacklistCNId = 13L;
        Set<Long> nonAvailableWorkerId = ImmutableSet.of(deadBEId, deadCNId, inBlacklistBEId, inBlacklistCNId);
        id2Backend.get(deadBEId).setAlive(false);
        id2ComputeNode.get(deadCNId).setAlive(false);
        new MockUp<SimpleScheduler>() {
            @Mock
            public boolean isInBlocklist(long backendId) {
                return backendId == inBlacklistBEId || backendId == inBlacklistCNId;
            }
        };

        Reference<Integer> nextComputeNodeIndex = new Reference<>(0);
        new MockUp<DefaultWorkerProvider>() {
            @Mock
            int getNextComputeNodeIndex() {
                int next = nextComputeNodeIndex.getRef();
                nextComputeNodeIndex.setRef(next + 1);
                return next;
            }
        };

        new MockUp<SystemInfoService>() {
            @Mock
            public ImmutableMap<Long, ComputeNode> getIdToBackend() {
                return id2Backend;
            }

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

        DefaultWorkerProvider.Factory workerProviderFactory = new DefaultWorkerProvider.Factory();
        DefaultWorkerProvider workerProvider;
        List<Integer> numUsedComputeNodesList = ImmutableList.of(100, 0, -1, 1, 2, 3, 4, 5, 6);
        for (Integer numUsedComputeNodes : numUsedComputeNodesList) {
            // Reset nextComputeNodeIndex.
            nextComputeNodeIndex.setRef(0);

            workerProvider =
                    workerProviderFactory.captureAvailableWorkers(GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo(),
                            true, numUsedComputeNodes, ComputationFragmentSchedulingPolicy.COMPUTE_NODES_ONLY, 
                            WarehouseManager.DEFAULT_WAREHOUSE_ID);

            int numAvailableComputeNodes = 0;
            for (long id = 0; id < 15; id++) {
                ComputeNode worker = workerProvider.getWorkerById(id);
                if (nonAvailableWorkerId.contains(id)
                        // Exceed the limitation of numUsedComputeNodes.
                        || (numUsedComputeNodes > 0 && numAvailableComputeNodes >= numUsedComputeNodes)) {
                    Assert.assertNull(worker);
                } else {
                    Assert.assertNotNull("numUsedComputeNodes=" + numUsedComputeNodes + ",id=" + id, worker);
                    Assert.assertEquals(id, worker.getId());

                    if (id2ComputeNode.containsKey(id)) {
                        numAvailableComputeNodes++;
                    }
                }
            }
        }
    }