@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 workerAddrUpdateWithIdUnchanged() throws Exception {
    KetamaHashPolicy policy = new KetamaHashPolicy(mConf);
    List<WorkerInfo> workers = new ArrayList<>();
    workers.add(new WorkerInfo().setIdentity(WorkerIdentityTestUtils.ofLegacyId(1L))
        .setAddress(new WorkerNetAddress().setHost("host1"))
        .setCapacityBytes(0)
        .setUsedBytes(0)
        .setState(WorkerState.LIVE));
    workers.add(new WorkerInfo().setIdentity(WorkerIdentityTestUtils.ofLegacyId(2L))
        .setAddress(new WorkerNetAddress().setHost("host2"))
        .setCapacityBytes(0)
        .setUsedBytes(0)
        .setState(WorkerState.LIVE));
    List<BlockWorkerInfo> selectedWorkers =
        policy.getPreferredWorkers(new WorkerClusterView(workers), "fileId", 2);
    assertEquals("host1",
        selectedWorkers.stream()
            .filter(w -> w.getIdentity().equals(WorkerIdentityTestUtils.ofLegacyId(1L)))
            .findFirst()
            .get()
            .getNetAddress()
            .getHost());

    // now the worker 1 has migrated to host 3
    workers.set(0, new WorkerInfo().setIdentity(WorkerIdentityTestUtils.ofLegacyId(1L))
        .setAddress(new WorkerNetAddress().setHost("host3"))
        .setCapacityBytes(0)
        .setUsedBytes(0)
        .setState(WorkerState.LIVE));
    List<BlockWorkerInfo> updatedWorkers =
        policy.getPreferredWorkers(new WorkerClusterView(workers), "fileId", 2);
    assertEquals(
        selectedWorkers.stream().map(BlockWorkerInfo::getIdentity).collect(Collectors.toList()),
        updatedWorkers.stream().map(BlockWorkerInfo::getIdentity).collect(Collectors.toList()));
    assertEquals("host3",
        updatedWorkers.stream()
            .filter(w -> w.getIdentity().equals(WorkerIdentityTestUtils.ofLegacyId(1L)))
            .findFirst()
            .get()
            .getNetAddress()
            .getHost());
  }