@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()
        .stream()
        .map(w -> new BlockWorkerInfo(
            w.getIdentity(),
            w.getAddress(),
            w.getCapacityBytes(),
            w.getUsedBytes()))
        .collect(Collectors.toList());
    BlockWorkerInfo workerWithMostBlocks = JobUtils.getWorkerWithMostBlocks(alluxioWorkerInfoList,
        context.getFileSystem().getStatus(uri).getFileBlockInfos());

    // Map the best Alluxio worker to a job worker.
    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;
  }
@Test
  public void selectExecutorsTest() throws Exception {
    AlluxioURI uri = new AlluxioURI("/test");
    PersistConfig config = new PersistConfig(uri.getPath(), -1, true, "");

    WorkerNetAddress workerNetAddress1 = new WorkerNetAddress().setDataPort(10).setHost("host1");
    WorkerIdentity workerIdentity1 = WorkerIdentityTestUtils.randomLegacyId();
    WorkerNetAddress workerNetAddress2 = new WorkerNetAddress().setDataPort(100).setHost("host2");
    WorkerIdentity workerIdentity2 = WorkerIdentityTestUtils.randomLegacyId();

    WorkerInfo blockWorkerInfo1 =
        new WorkerInfo().setIdentity(workerIdentity1).setAddress(workerNetAddress1)
            .setCapacityBytes(1).setUsedBytes(1);
    WorkerInfo blockWorkerInfo2 =
        new WorkerInfo().setIdentity(workerIdentity2).setAddress(workerNetAddress2)
            .setCapacityBytes(1).setUsedBytes(1);

    WorkerInfo workerInfo1 = new WorkerInfo()
        .setIdentity(workerIdentity1)
        .setAddress(workerNetAddress1);
    WorkerInfo workerInfo2 = new WorkerInfo()
        .setIdentity(workerIdentity2)
        .setAddress(workerNetAddress2);

    FileBlockInfo fileBlockInfo1 = mockFileBlockInfo(1, workerNetAddress2);
    FileBlockInfo fileBlockInfo2 = mockFileBlockInfo(2, workerNetAddress1);
    FileBlockInfo fileBlockInfo3 = mockFileBlockInfo(3, workerNetAddress1);
    FileInfo testFileInfo = new FileInfo();
    testFileInfo.setFileBlockInfos(
        Lists.newArrayList(fileBlockInfo1, fileBlockInfo2, fileBlockInfo3));

    Mockito.when(mMockFileSystemContext.getCachedWorkers()).thenReturn(
        new WorkerClusterView(Arrays.asList(blockWorkerInfo1, blockWorkerInfo2)));
    Mockito.when(mMockFileSystem.getStatus(uri)).thenReturn(new URIStatus(testFileInfo));

    Set<Pair<WorkerInfo, SerializableVoid>> result = new PersistDefinition()
        .selectExecutors(config, Lists.newArrayList(workerInfo2, workerInfo1),
            new SelectExecutorsContext(1, mJobServerContext));
    Assert.assertEquals(1, result.size());
    Assert.assertEquals(workerInfo1, result.iterator().next().getFirst());
  }