@Override
  public List<BlockWorkerInfo> getPreferredWorkers(WorkerClusterView workerClusterView,
      String fileId, int count) throws ResourceExhaustedException {
    if (workerClusterView.size() < count) {
      throw new ResourceExhaustedException(String.format(
          "Not enough workers in the cluster %d workers in the cluster but %d required",
          workerClusterView.size(), count));
    }
    Set<WorkerIdentity> workerIdentities = workerClusterView.workerIds();
    mHashProvider.refresh(workerIdentities);
    List<WorkerIdentity> workers = mHashProvider.getMultiple(fileId, count);
    if (workers.size() != count) {
      throw new ResourceExhaustedException(String.format(
          "Found %d workers from the hash ring but %d required", workers.size(), count));
    }
    ImmutableList.Builder<BlockWorkerInfo> builder = ImmutableList.builder();
    for (WorkerIdentity worker : workers) {
      Optional<WorkerInfo> optionalWorkerInfo = workerClusterView.getWorkerById(worker);
      final WorkerInfo workerInfo;
      if (optionalWorkerInfo.isPresent()) {
        workerInfo = optionalWorkerInfo.get();
      } else {
        // the worker returned by the policy does not exist in the cluster view
        // supplied by the client.
        // this can happen when the membership changes and some callers fail to update
        // to the latest worker cluster view.
        // in this case, just skip this worker
        LOG.debug("Inconsistency between caller's view of cluster and that of "
            + "the consistent hash policy's: worker {} selected by policy does not exist in "
            + "caller's view {}. Skipping this worker.",
            worker, workerClusterView);
        continue;
      }

      BlockWorkerInfo blockWorkerInfo = new BlockWorkerInfo(
          worker, workerInfo.getAddress(), workerInfo.getCapacityBytes(),
          workerInfo.getUsedBytes(), workerInfo.getState() == WorkerState.LIVE
      );
      builder.add(blockWorkerInfo);
    }
    List<BlockWorkerInfo> infos = builder.build();
    return infos;
  }
@Test
  public void getMultipleWorkers() throws Exception {
    WorkerLocationPolicy policy = WorkerLocationPolicy.Factory.create(mConf);
    assertTrue(policy instanceof ConsistentHashPolicy);
    // Prepare a worker list
    WorkerClusterView workers = new WorkerClusterView(Arrays.asList(
        new WorkerInfo()
            .setIdentity(WorkerIdentityTestUtils.ofLegacyId(1))
            .setAddress(new WorkerNetAddress()
                .setHost("master1").setRpcPort(29998).setDataPort(29999).setWebPort(30000))
            .setCapacityBytes(1024)
            .setUsedBytes(0),
        new WorkerInfo()
            .setIdentity(WorkerIdentityTestUtils.ofLegacyId(2))
            .setAddress(new WorkerNetAddress()
                .setHost("master2").setRpcPort(29998).setDataPort(29999).setWebPort(30000))
            .setCapacityBytes(1024)
            .setUsedBytes(0)));

    List<BlockWorkerInfo> assignedWorkers = policy.getPreferredWorkers(workers, "hdfs://a/b/c", 2);
    assertEquals(2, assignedWorkers.size());
    assertTrue(assignedWorkers.stream().allMatch(w -> contains(workers, w)));
    // The order of the workers should be consistent
    assertEquals(assignedWorkers.get(0).getNetAddress().getHost(), "master2");
    assertEquals(assignedWorkers.get(1).getNetAddress().getHost(), "master1");
    assertThrows(ResourceExhaustedException.class, () -> {
      // Getting 2 out of 1 worker will result in an error
      policy.getPreferredWorkers(
          new WorkerClusterView(Arrays.asList(
              new WorkerInfo()
                  .setIdentity(WorkerIdentityTestUtils.ofLegacyId(1))
                  .setAddress(new WorkerNetAddress()
                      .setHost("master1").setRpcPort(29998).setDataPort(29999).setWebPort(30000))
                  .setCapacityBytes(1024)
                  .setUsedBytes(0))),
          "hdfs://a/b/c", 2);
    });
  }