@Nullable
    public static TNetworkAddress getHost(long nodeId,
                                          List<TScanRangeLocation> locations,
                                          ImmutableMap<Long, ComputeNode> computeNodes,
                                          Reference<Long> nodeIdRef) {

        if (locations == null || computeNodes == null) {
            return null;
        }
        LOG.debug("getHost nodeID={}, nodeSize={}", nodeId, computeNodes.size());

        ComputeNode node = computeNodes.get(nodeId);
        if (node != null && node.isAlive() && !HOST_BLACKLIST.contains(nodeId)) {
            nodeIdRef.setRef(nodeId);
            return new TNetworkAddress(node.getHost(), node.getBePort());
        } else {
            for (TScanRangeLocation location : locations) {
                if (location.backend_id == nodeId) {
                    continue;
                }
                // choose the first alive backend(in analysis stage, the locations are random)
                ComputeNode candidateNode = computeNodes.get(location.backend_id);
                if (candidateNode != null && candidateNode.isAlive()
                        && !HOST_BLACKLIST.contains(location.backend_id)) {
                    nodeIdRef.setRef(location.backend_id);
                    return new TNetworkAddress(candidateNode.getHost(), candidateNode.getBePort());
                }
            }

            // In shared data mode, we can select any alive node to replace the original dead node for query
            if (RunMode.isSharedDataMode()) {
                List<ComputeNode> allNodes = new ArrayList<>(computeNodes.size());
                allNodes.addAll(computeNodes.values());
                List<ComputeNode> candidateNodes = allNodes.stream()
                        .filter(x -> x.getId() != nodeId && x.isAlive() &&
                                !HOST_BLACKLIST.contains(x.getId())).collect(Collectors.toList());
                if (!candidateNodes.isEmpty()) {
                    // use modulo operation to ensure that the same node is selected for the dead node
                    ComputeNode candidateNode = candidateNodes.get((int) (nodeId % candidateNodes.size()));
                    nodeIdRef.setRef(candidateNode.getId());
                    return new TNetworkAddress(candidateNode.getHost(), candidateNode.getBePort());
                }
            }
        }

        // no backend or compute node returned
        return null;
    }
@Test
    public void testGetHostWithBackendIdInSharedDataMode() {
        // locations
        List<TScanRangeLocation> locations = new ArrayList<TScanRangeLocation>();
        TScanRangeLocation locationA = new TScanRangeLocation();
        locationA.setBackend_id(0);
        locations.add(locationA);

        // backends
        Backend backendA = new Backend(0, "addressA", 0);
        backendA.updateOnce(0, 0, 0);
        Backend backendB = new Backend(1, "addressB", 0);
        backendB.updateOnce(0, 0, 0);
        Backend backendC = new Backend(2, "addressC", 0);
        backendC.updateOnce(0, 0, 0);

        Map<Long, Backend> backends = Maps.newHashMap();
        backends.put((long) 0, backendA);
        backends.put((long) 1, backendB);
        backends.put((long) 2, backendC);

        new MockUp<RunMode>() {
            @Mock
            public RunMode getCurrentRunMode() {
                return RunMode.SHARED_DATA;
            }
        };

        ImmutableMap<Long, ComputeNode> immutableBackends = null;

        {
            // backendA in locations is alive
            backendA.setAlive(true);
            backendB.setAlive(true);
            backendC.setAlive(true);
            immutableBackends = ImmutableMap.copyOf(backends);

            Assert.assertEquals(
                    SimpleScheduler.getHost(0, locations, immutableBackends, ref).hostname,
                    "addressA");
        }

        {
            // backendA in locations is not alive
            backendA.setAlive(false);
            backendB.setAlive(false);
            backendC.setAlive(true);
            immutableBackends = ImmutableMap.copyOf(backends);

            Assert.assertEquals(
                    SimpleScheduler.getHost(0, locations, immutableBackends, ref).hostname,
                    "addressC");
        }
    }