@Test public void chooseClosestTierAvoidEviction() throws Exception { List<BlockWorkerInfo> workers = new ArrayList<>(); workers.add(worker(Constants.GB, Constants.MB, "node2", "rack3")); workers.add(worker(Constants.GB, 0, "node3", "rack2")); workers.add(worker(Constants.GB, 0, "node4", "rack3")); BlockLocationPolicy policy; WorkerNetAddress chosen; policy = new LocalFirstAvoidEvictionPolicy( mConf.getBytes(PropertyKey.USER_BLOCK_AVOID_EVICTION_POLICY_RESERVED_BYTES), TieredIdentityFactory.fromString("node=node2,rack=rack3", mConf), mConf); GetWorkerOptions options = GetWorkerOptions.defaults() .setBlockWorkerInfos(workers).setBlockInfo(new BlockInfo().setLength(Constants.GB)); chosen = policy.getWorker(options); assertEquals("node4", chosen.getTieredIdentity().getTier(0).getValue()); }
@Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> allWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); List<BlockWorkerInfo> workers = allWorkers.stream() .filter(worker -> getAvailableBytes(worker) >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); if (workers.isEmpty()) { workers = allWorkers; } GetWorkerOptions filteredWorkers = GetWorkerOptions.defaults() .setBlockInfo(options.getBlockInfo()) .setBlockWorkerInfos(workers); return mPolicy.getWorker(filteredWorkers); }
LocalFirstAvoidEvictionPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> allWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); List<BlockWorkerInfo> workers = allWorkers.stream() .filter(worker -> getAvailableBytes(worker) >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); if (workers.isEmpty()) { workers = allWorkers; } GetWorkerOptions filteredWorkers = GetWorkerOptions.defaults() .setBlockInfo(options.getBlockInfo()) .setBlockWorkerInfos(workers); return mPolicy.getWorker(filteredWorkers); } }
LocalFirstAvoidEvictionPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> allWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); List<BlockWorkerInfo> workers = allWorkers.stream() .filter(worker -> getAvailableBytes(worker) >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); if (workers.isEmpty()) { workers = allWorkers; } GetWorkerOptions filteredWorkers = GetWorkerOptions.defaults() .setBlockInfo(options.getBlockInfo()) .setBlockWorkerInfos(workers); return mPolicy.getWorker(filteredWorkers); } LocalFirstAvoidEvictionPolicy(AlluxioConfiguration conf);   LocalFirstAvoidEvictionPolicy(long blockCapacityReserved,
      AlluxioConfiguration conf); @VisibleForTesting  LocalFirstAvoidEvictionPolicy(long blockCapacityReserved, TieredIdentity identity,
      AlluxioConfiguration conf); }
LocalFirstAvoidEvictionPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> allWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); List<BlockWorkerInfo> workers = allWorkers.stream() .filter(worker -> getAvailableBytes(worker) >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); if (workers.isEmpty()) { workers = allWorkers; } GetWorkerOptions filteredWorkers = GetWorkerOptions.defaults() .setBlockInfo(options.getBlockInfo()) .setBlockWorkerInfos(workers); return mPolicy.getWorker(filteredWorkers); } LocalFirstAvoidEvictionPolicy(AlluxioConfiguration conf);   LocalFirstAvoidEvictionPolicy(long blockCapacityReserved,
      AlluxioConfiguration conf); @VisibleForTesting  LocalFirstAvoidEvictionPolicy(long blockCapacityReserved, TieredIdentity identity,
      AlluxioConfiguration conf); @Override WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); }
LocalFirstAvoidEvictionPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> allWorkers = Lists.newArrayList(options.getBlockWorkerInfos()); List<BlockWorkerInfo> workers = allWorkers.stream() .filter(worker -> getAvailableBytes(worker) >= options.getBlockInfo().getLength()) .collect(Collectors.toList()); if (workers.isEmpty()) { workers = allWorkers; } GetWorkerOptions filteredWorkers = GetWorkerOptions.defaults() .setBlockInfo(options.getBlockInfo()) .setBlockWorkerInfos(workers); return mPolicy.getWorker(filteredWorkers); } LocalFirstAvoidEvictionPolicy(AlluxioConfiguration conf);   LocalFirstAvoidEvictionPolicy(long blockCapacityReserved,
      AlluxioConfiguration conf); @VisibleForTesting  LocalFirstAvoidEvictionPolicy(long blockCapacityReserved, TieredIdentity identity,
      AlluxioConfiguration conf); @Override WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }