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

        int bucketNum = 3;
        OlapScanNode scanNode = newOlapScanNode(1, bucketNum);
        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();
            NormalBackendSelector selector =
                    new NormalBackendSelector(scanNode, scanLocations, assignment, provider, false);
            // 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();
            NormalBackendSelector selector =
                    new NormalBackendSelector(scanNode, scanLocations, assignment, provider1, false);
            // 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();
            NormalBackendSelector selector =
                    new NormalBackendSelector(scanNode, scanLocations, assignment, providerNoAvailNode, false);
            Assert.assertThrows(NonRecoverableException.class, selector::computeScanRangeAssignment);
        }
    }