@Test public void getOthersWhenNotEnoughCapacityOnLocal() { String localhostName = NetworkAddressUtils.getLocalHostName(sResolutionTimeout); LocalFirstPolicy policy = new LocalFirstPolicy(sConf); List<BlockWorkerInfo> workers = new ArrayList<>(); workers.add(worker(Constants.GB, "worker1", "")); workers.add(worker(Constants.MB, localhostName, "")); GetWorkerOptions options = GetWorkerOptions.defaults() .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.GB)); assertEquals("worker1", policy.getWorker(options).getHost()); }
@Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> shuffledWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(shuffledWorkers); List<BlockWorkerInfo> candidateWorkers = shuffledWorkers.stream() .filter(worker -> worker.getCapacityBytes() >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); List<WorkerNetAddress> addresses = candidateWorkers.stream() .map(worker -> worker.getNetAddress()) .filter(Objects::nonNull) .collect(Collectors.toList()); Optional<Pair<WorkerNetAddress, Boolean>> nearest = BlockLocationUtils.nearest(mTieredIdentity, addresses, mConf); if (!nearest.isPresent()) { return null; } return nearest.get().getFirst(); }
LocalFirstPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> shuffledWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(shuffledWorkers); List<BlockWorkerInfo> candidateWorkers = shuffledWorkers.stream() .filter(worker -> worker.getCapacityBytes() >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); List<WorkerNetAddress> addresses = candidateWorkers.stream() .map(worker -> worker.getNetAddress()) .filter(Objects::nonNull) .collect(Collectors.toList()); Optional<Pair<WorkerNetAddress, Boolean>> nearest = BlockLocationUtils.nearest(mTieredIdentity, addresses, mConf); if (!nearest.isPresent()) { return null; } return nearest.get().getFirst(); } }
LocalFirstPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> shuffledWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(shuffledWorkers); List<BlockWorkerInfo> candidateWorkers = shuffledWorkers.stream() .filter(worker -> worker.getCapacityBytes() >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); List<WorkerNetAddress> addresses = candidateWorkers.stream() .map(worker -> worker.getNetAddress()) .filter(Objects::nonNull) .collect(Collectors.toList()); Optional<Pair<WorkerNetAddress, Boolean>> nearest = BlockLocationUtils.nearest(mTieredIdentity, addresses, mConf); if (!nearest.isPresent()) { return null; } return nearest.get().getFirst(); } LocalFirstPolicy(AlluxioConfiguration conf); @VisibleForTesting  LocalFirstPolicy(TieredIdentity identity, AlluxioConfiguration conf); }
LocalFirstPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> shuffledWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(shuffledWorkers); List<BlockWorkerInfo> candidateWorkers = shuffledWorkers.stream() .filter(worker -> worker.getCapacityBytes() >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); List<WorkerNetAddress> addresses = candidateWorkers.stream() .map(worker -> worker.getNetAddress()) .filter(Objects::nonNull) .collect(Collectors.toList()); Optional<Pair<WorkerNetAddress, Boolean>> nearest = BlockLocationUtils.nearest(mTieredIdentity, addresses, mConf); if (!nearest.isPresent()) { return null; } return nearest.get().getFirst(); } LocalFirstPolicy(AlluxioConfiguration conf); @VisibleForTesting  LocalFirstPolicy(TieredIdentity identity, AlluxioConfiguration conf); @Override WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); }
LocalFirstPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> shuffledWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(shuffledWorkers); List<BlockWorkerInfo> candidateWorkers = shuffledWorkers.stream() .filter(worker -> worker.getCapacityBytes() >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); List<WorkerNetAddress> addresses = candidateWorkers.stream() .map(worker -> worker.getNetAddress()) .filter(Objects::nonNull) .collect(Collectors.toList()); Optional<Pair<WorkerNetAddress, Boolean>> nearest = BlockLocationUtils.nearest(mTieredIdentity, addresses, mConf); if (!nearest.isPresent()) { return null; } return nearest.get().getFirst(); } LocalFirstPolicy(AlluxioConfiguration conf); @VisibleForTesting  LocalFirstPolicy(TieredIdentity identity, AlluxioConfiguration conf); @Override WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }