@Override
  public List<BlockWorkerInfo> getPreferredWorkers(WorkerClusterView workerClusterView,
      String fileId, int count) throws ResourceExhaustedException {
    String userHostname = NetworkAddressUtils.getClientHostName(mConf);
    // TODO(jiacheng): domain socket is not considered here
    // Find the worker matching in hostname
    List<BlockWorkerInfo> results = new ArrayList<>();
    for (WorkerInfo worker : workerClusterView) {
      WorkerNetAddress workerAddr = worker.getAddress();
      if (workerAddr == null) {
        continue;
      }
      // Only a plain string match is performed on hostname
      // If one is IP and the other is hostname, a false negative will be returned
      if (userHostname.equals(workerAddr.getHost())) {
        results.add(new BlockWorkerInfo(worker.getIdentity(),
            workerAddr, worker.getCapacityBytes(), worker.getUsedBytes(),
            worker.getState() == WorkerState.LIVE));
        if (results.size() >= count) {
          break;
        }
      }
    }
    if (results.size() < count) {
      throw new ResourceExhaustedException(String.format(
          "Failed to find a local worker for client hostname %s", userHostname));
    }
    return results;
  }
@Test
  public void getOneWorker() throws Exception {
    WorkerLocationPolicy policy = WorkerLocationPolicy.Factory.create(mConf);
    assertTrue(policy instanceof LocalWorkerPolicy);
    // Prepare a worker list
    WorkerClusterView workers = new WorkerClusterView(Arrays.asList(
        new WorkerInfo()
            .setIdentity(LOCAL_WORKER_ID)
            .setAddress(new WorkerNetAddress()
                .setHost(LOCAL_HOSTNAME).setRpcPort(29998).setDataPort(29999).setWebPort(30000))
            .setCapacityBytes(1024)
            .setUsedBytes(0),
        new WorkerInfo()
            .setIdentity(WorkerIdentityTestUtils.randomLegacyId())
            .setAddress(new WorkerNetAddress()
                .setHost("remotehost").setRpcPort(29998).setDataPort(29999).setWebPort(30000))
            .setCapacityBytes(1024)
            .setUsedBytes(0)));

    List<BlockWorkerInfo> assignedWorkers = policy.getPreferredWorkers(workers, "hdfs://a/b/c", 1);
    assertEquals(1, assignedWorkers.size());
    assertTrue(contains(workers, assignedWorkers.get(0)));
    assertEquals(LOCAL_WORKER_ID, assignedWorkers.get(0).getIdentity());
    assertEquals(LOCAL_HOSTNAME, assignedWorkers.get(0).getNetAddress().getHost());

    assertThrows(ResourceExhaustedException.class, () -> {
      // Getting 1 out of no workers will result in an error
      policy.getPreferredWorkers(new WorkerClusterView(ImmutableList.of()), "hdfs://a/b/c", 1);
    });
  }