@Override
    public long selectNextWorker() throws NonRecoverableException {
        ComputeNode worker;
        worker = getNextWorker(availableID2ComputeNode, DefaultSharedDataWorkerProvider::getNextComputeNodeIndex);

        if (worker == null) {
            reportWorkerNotFoundException();
        }
        Preconditions.checkNotNull(worker);
        selectWorkerUnchecked(worker.getId());
        return worker.getId();
    }
@Test
    public void testSelectNextWorker() throws UserException {
        HostBlacklist blockList = SimpleScheduler.getHostBlacklist();
        blockList.hostBlacklist.clear();
        SystemInfoService sysInfo = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo();

        List<Long> availList = prepareNodeAliveAndBlock(sysInfo, blockList);
        { // test backend nodes only
            ImmutableMap.Builder<Long, ComputeNode> builder = new ImmutableMap.Builder<>();
            for (long backendId : id2Backend.keySet()) {
                if (availList.contains(backendId)) {
                    builder.put(backendId, id2Backend.get(backendId));
                }
            }
            ImmutableMap<Long, ComputeNode> availableId2ComputeNode = builder.build();
            WorkerProvider workerProvider =
                    new DefaultSharedDataWorkerProvider(ImmutableMap.copyOf(id2Backend), availableId2ComputeNode);
            testSelectNextWorkerHelper(workerProvider, availableId2ComputeNode);
        }

        { // test compute nodes only
            ImmutableMap.Builder<Long, ComputeNode> builder = new ImmutableMap.Builder<>();
            for (long backendId : id2ComputeNode.keySet()) {
                if (availList.contains(backendId)) {
                    builder.put(backendId, id2ComputeNode.get(backendId));
                }
            }
            ImmutableMap<Long, ComputeNode> availableId2ComputeNode = builder.build();
            WorkerProvider workerProvider =
                    new DefaultSharedDataWorkerProvider(ImmutableMap.copyOf(id2ComputeNode), availableId2ComputeNode);
            testSelectNextWorkerHelper(workerProvider, availableId2ComputeNode);
        }

        { // test both backends and compute nodes
            ImmutableMap.Builder<Long, ComputeNode> builder = new ImmutableMap.Builder<>();
            for (long backendId : id2AllNodes.keySet()) {
                if (availList.contains(backendId)) {
                    builder.put(backendId, id2AllNodes.get(backendId));
                }
            }
            ImmutableMap<Long, ComputeNode> availableId2ComputeNode = builder.build();
            WorkerProvider workerProvider =
                    new DefaultSharedDataWorkerProvider(ImmutableMap.copyOf(id2AllNodes), availableId2ComputeNode);
            testSelectNextWorkerHelper(workerProvider, availableId2ComputeNode);
        }

        { // test no available worker to select
            WorkerProvider workerProvider =
                    new DefaultSharedDataWorkerProvider(ImmutableMap.copyOf(id2AllNodes), ImmutableMap.of());
            Assert.assertThrows(NonRecoverableException.class, workerProvider::selectNextWorker);
        }
    }