@Override
    public boolean isDataNodeAvailable(long dataNodeId) {
        // DataNode and ComputeNode is exchangeable in SHARED_DATA mode
        return availableID2ComputeNode.containsKey(dataNodeId);
    }
@Test
    public void testCollocationBackendSelectorWithSharedDataWorkerProvider() {
        HostBlacklist blockList = SimpleScheduler.getHostBlacklist();
        blockList.hostBlacklist.clear();
        SystemInfoService sysInfo = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo();
        List<Long> availList = prepareNodeAliveAndBlock(sysInfo, blockList);

        int bucketNum = 6;
        OlapScanNode scanNode = newOlapScanNode(10, bucketNum);
        final Map<Integer, List<Long>> bucketSeqToBackends = ImmutableMap.of(
                0, ImmutableList.of(1L),
                1, ImmutableList.of(2L),
                2, ImmutableList.of(3L),
                3, ImmutableList.of(4L),
                4, ImmutableList.of(5L),
                5, ImmutableList.of(6L)
        );
        scanNode.bucketSeq2locations = genBucketSeq2Locations(bucketSeqToBackends, 3);
        List<TScanRangeLocations> scanLocations = generateScanRangeLocations(id2AllNodes, 10, bucketNum);
        WorkerProvider provider = newWorkerProvider();

        int nonAvailNum = 0;
        for (TScanRangeLocations locations : scanLocations) {
            for (TScanRangeLocation location : locations.getLocations()) {
                if (!provider.isDataNodeAvailable(location.getBackend_id())) {
                    ++nonAvailNum;
                }
            }
        }
        // the scanRangeLocations contains non-avail locations
        Assert.assertTrue(nonAvailNum > 0);

        { // normal case
            FragmentScanRangeAssignment assignment = new FragmentScanRangeAssignment();
            ColocatedBackendSelector.Assignment colAssignment = new ColocatedBackendSelector.Assignment(scanNode, 1);
            ColocatedBackendSelector selector =
                    new ColocatedBackendSelector(scanNode, assignment, colAssignment, false, provider, 1);
            // the computation will not fail even though there are non-available locations
            ExceptionChecker.expectThrowsNoException(selector::computeScanRangeAssignment);

            // check the assignment, should be all in the availList
            for (long id : assignment.keySet()) {
                Assert.assertTrue(availList.contains(id));
            }
        }

        { // make only one node available, the final assignment will be all on the single available node
            ComputeNode availNode = id2AllNodes.get(availList.get(0));
            WorkerProvider provider1 = new DefaultSharedDataWorkerProvider(ImmutableMap.copyOf(id2AllNodes),
                    ImmutableMap.of(availNode.getId(), availNode));

            FragmentScanRangeAssignment assignment = new FragmentScanRangeAssignment();
            ColocatedBackendSelector.Assignment colAssignment = new ColocatedBackendSelector.Assignment(scanNode, 1);
            ColocatedBackendSelector selector =
                    new ColocatedBackendSelector(scanNode, assignment, colAssignment, false, provider1, 1);
            // the computation will not fail even though there are non-available locations
            ExceptionChecker.expectThrowsNoException(selector::computeScanRangeAssignment);

            Assert.assertEquals(1, assignment.size());
            // check the assignment, should be all in the availList
            for (long id : assignment.keySet()) {
                Assert.assertEquals(availNode.getId(), id);
            }
        }

        { // make no node available. Exception throws
            WorkerProvider providerNoAvailNode = new DefaultSharedDataWorkerProvider(ImmutableMap.copyOf(id2AllNodes),
                    ImmutableMap.of());
            FragmentScanRangeAssignment assignment = new FragmentScanRangeAssignment();
            ColocatedBackendSelector.Assignment colAssignment = new ColocatedBackendSelector.Assignment(scanNode, 1);
            ColocatedBackendSelector selector =
                    new ColocatedBackendSelector(scanNode, assignment, colAssignment, false, providerNoAvailNode, 1);
            Assert.assertThrows(NonRecoverableException.class, selector::computeScanRangeAssignment);
        }
    }