@Test public void selectExecutorsTest() throws Exception { AlluxioURI uri = new AlluxioURI("/test"); PersistConfig config = new PersistConfig(uri.getPath(), -1, true, ""); WorkerNetAddress workerNetAddress = new WorkerNetAddress().setDataPort(10); WorkerInfo workerInfo = new WorkerInfo().setAddress(workerNetAddress); long blockId = 1; BlockInfo blockInfo = new BlockInfo().setBlockId(blockId); FileBlockInfo fileBlockInfo = new FileBlockInfo().setBlockInfo(blockInfo); BlockLocation location = new BlockLocation(); location.setWorkerAddress(workerNetAddress); blockInfo.setLocations(Lists.newArrayList(location)); FileInfo testFileInfo = new FileInfo(); testFileInfo.setFileBlockInfos(Lists.newArrayList(fileBlockInfo)); Mockito.when(mMockFileSystem.getStatus(uri)).thenReturn(new URIStatus(testFileInfo)); Set<Pair<WorkerInfo, SerializableVoid>> result = new PersistDefinition().selectExecutors(config, Lists.newArrayList(workerInfo), new SelectExecutorsContext(1, mJobServerContext)); Assert.assertEquals(1, result.size()); Assert.assertEquals(workerInfo, result.iterator().next().getFirst()); }
@Override public Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { if (jobWorkerInfoList.isEmpty()) { throw new RuntimeException("No worker is available"); } AlluxioURI uri = new AlluxioURI(config.getFilePath()); List<BlockWorkerInfo> alluxioWorkerInfoList = context.getFsContext().getCachedWorkers(); BlockWorkerInfo workerWithMostBlocks = JobUtils.getWorkerWithMostBlocks(alluxioWorkerInfoList, context.getFileSystem().getStatus(uri).getFileBlockInfos()); Set<Pair<WorkerInfo, SerializableVoid>> result = Sets.newHashSet(); boolean found = false; if (workerWithMostBlocks != null) { for (WorkerInfo workerInfo : jobWorkerInfoList) { if (workerInfo.getAddress().getHost() .equals(workerWithMostBlocks.getNetAddress().getHost())) { result.add(new Pair<>(workerInfo, null)); found = true; break; } } } if (!found) { result.add(new Pair<>( jobWorkerInfoList.get(new Random().nextInt(jobWorkerInfoList.size())), null)); } return result; }
PersistDefinition extends AbstractVoidPlanDefinition<PersistConfig, SerializableVoid> { @Override public Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { if (jobWorkerInfoList.isEmpty()) { throw new RuntimeException("No worker is available"); } AlluxioURI uri = new AlluxioURI(config.getFilePath()); List<BlockWorkerInfo> alluxioWorkerInfoList = context.getFsContext().getCachedWorkers(); BlockWorkerInfo workerWithMostBlocks = JobUtils.getWorkerWithMostBlocks(alluxioWorkerInfoList, context.getFileSystem().getStatus(uri).getFileBlockInfos()); Set<Pair<WorkerInfo, SerializableVoid>> result = Sets.newHashSet(); boolean found = false; if (workerWithMostBlocks != null) { for (WorkerInfo workerInfo : jobWorkerInfoList) { if (workerInfo.getAddress().getHost() .equals(workerWithMostBlocks.getNetAddress().getHost())) { result.add(new Pair<>(workerInfo, null)); found = true; break; } } } if (!found) { result.add(new Pair<>( jobWorkerInfoList.get(new Random().nextInt(jobWorkerInfoList.size())), null)); } return result; } }
PersistDefinition extends AbstractVoidPlanDefinition<PersistConfig, SerializableVoid> { @Override public Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { if (jobWorkerInfoList.isEmpty()) { throw new RuntimeException("No worker is available"); } AlluxioURI uri = new AlluxioURI(config.getFilePath()); List<BlockWorkerInfo> alluxioWorkerInfoList = context.getFsContext().getCachedWorkers(); BlockWorkerInfo workerWithMostBlocks = JobUtils.getWorkerWithMostBlocks(alluxioWorkerInfoList, context.getFileSystem().getStatus(uri).getFileBlockInfos()); Set<Pair<WorkerInfo, SerializableVoid>> result = Sets.newHashSet(); boolean found = false; if (workerWithMostBlocks != null) { for (WorkerInfo workerInfo : jobWorkerInfoList) { if (workerInfo.getAddress().getHost() .equals(workerWithMostBlocks.getNetAddress().getHost())) { result.add(new Pair<>(workerInfo, null)); found = true; break; } } } if (!found) { result.add(new Pair<>( jobWorkerInfoList.get(new Random().nextInt(jobWorkerInfoList.size())), null)); } return result; } PersistDefinition(); }
PersistDefinition extends AbstractVoidPlanDefinition<PersistConfig, SerializableVoid> { @Override public Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { if (jobWorkerInfoList.isEmpty()) { throw new RuntimeException("No worker is available"); } AlluxioURI uri = new AlluxioURI(config.getFilePath()); List<BlockWorkerInfo> alluxioWorkerInfoList = context.getFsContext().getCachedWorkers(); BlockWorkerInfo workerWithMostBlocks = JobUtils.getWorkerWithMostBlocks(alluxioWorkerInfoList, context.getFileSystem().getStatus(uri).getFileBlockInfos()); Set<Pair<WorkerInfo, SerializableVoid>> result = Sets.newHashSet(); boolean found = false; if (workerWithMostBlocks != null) { for (WorkerInfo workerInfo : jobWorkerInfoList) { if (workerInfo.getAddress().getHost() .equals(workerWithMostBlocks.getNetAddress().getHost())) { result.add(new Pair<>(workerInfo, null)); found = true; break; } } } if (!found) { result.add(new Pair<>( jobWorkerInfoList.get(new Random().nextInt(jobWorkerInfoList.size())), null)); } return result; } PersistDefinition(); @Override Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config,
      List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context); @Override SerializableVoid runTask(PersistConfig config, SerializableVoid args,
      RunTaskContext context); @Override Class<PersistConfig> getJobConfigClass(); }
PersistDefinition extends AbstractVoidPlanDefinition<PersistConfig, SerializableVoid> { @Override public Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { if (jobWorkerInfoList.isEmpty()) { throw new RuntimeException("No worker is available"); } AlluxioURI uri = new AlluxioURI(config.getFilePath()); List<BlockWorkerInfo> alluxioWorkerInfoList = context.getFsContext().getCachedWorkers(); BlockWorkerInfo workerWithMostBlocks = JobUtils.getWorkerWithMostBlocks(alluxioWorkerInfoList, context.getFileSystem().getStatus(uri).getFileBlockInfos()); Set<Pair<WorkerInfo, SerializableVoid>> result = Sets.newHashSet(); boolean found = false; if (workerWithMostBlocks != null) { for (WorkerInfo workerInfo : jobWorkerInfoList) { if (workerInfo.getAddress().getHost() .equals(workerWithMostBlocks.getNetAddress().getHost())) { result.add(new Pair<>(workerInfo, null)); found = true; break; } } } if (!found) { result.add(new Pair<>( jobWorkerInfoList.get(new Random().nextInt(jobWorkerInfoList.size())), null)); } return result; } PersistDefinition(); @Override Set<Pair<WorkerInfo, SerializableVoid>> selectExecutors(PersistConfig config,
      List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context); @Override SerializableVoid runTask(PersistConfig config, SerializableVoid args,
      RunTaskContext context); @Override Class<PersistConfig> getJobConfigClass();  }