code: @Override
    @Nullable
    public WorkerNetAddress getWorker(GetWorkerOptions options) {
      Set<WorkerNetAddress> eligibleAddresses = new HashSet<>();
      for (BlockWorkerInfo info : options.getBlockWorkerInfos()) {
        eligibleAddresses.add(info.getNetAddress());
      }
  
      WorkerNetAddress address = mBlockLocationCache.get(options.getBlockInfo().getBlockId());
      if (address != null && eligibleAddresses.contains(address)) {
        return address;
      } else {
        address = null;
      }
  
      if (!mInitialized) {
        mWorkerInfoList = Lists.newArrayList(options.getBlockWorkerInfos());
        Collections.shuffle(mWorkerInfoList);
        mIndex = 0;
        mInitialized = true;
      }
  
      // at most try all the workers
      for (int i = 0; i < mWorkerInfoList.size(); i++) {
        WorkerNetAddress candidate = mWorkerInfoList.get(mIndex).getNetAddress();
        BlockWorkerInfo workerInfo = findBlockWorkerInfo(options.getBlockWorkerInfos(), candidate);
        mIndex = (mIndex + 1) % mWorkerInfoList.size();
        if (workerInfo != null
            && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength()
            && eligibleAddresses.contains(candidate)) {
          address = candidate;
          break;
        }
      }
      mBlockLocationCache.put(options.getBlockInfo().getBlockId(), address);
      return address;
    }
docs: The policy uses the first fetch of worker info list as the base, and visits each of them in a
  round-robin manner in the subsequent calls. The policy doesn't assume the list of worker info
  in the subsequent calls has the same order from the first, and it will skip the workers that
  are no longer active.
  
  @param options options
  @return the address of the worker to write to