alluxio-master-2.x/core/server/master/src/test/java/alluxio/master/file/scheduler/SchedulerTest.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.master.file.scheduler;

import static alluxio.master.file.scheduler.LoadTestUtils.fileWithBlockLocations;
import static alluxio.master.file.scheduler.LoadTestUtils.generateRandomBlockStatus;
import static alluxio.master.file.scheduler.LoadTestUtils.generateRandomFileInfo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import alluxio.Constants;
import alluxio.client.block.stream.BlockWorkerClient;
import alluxio.client.file.FileSystemContext;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.conf.Source;
import alluxio.exception.AccessControlException;
import alluxio.exception.runtime.ResourceExhaustedRuntimeException;
import alluxio.exception.status.UnavailableException;
import alluxio.grpc.BlockStatus;
import alluxio.grpc.JobProgressReportFormat;
import alluxio.grpc.LoadRequest;
import alluxio.grpc.LoadResponse;
import alluxio.grpc.TaskStatus;
import alluxio.job.JobDescription;
import alluxio.master.file.FileSystemMaster;
import alluxio.master.job.FileIterable;
import alluxio.master.job.LoadJob;
import alluxio.master.journal.JournalContext;
import alluxio.master.scheduler.DefaultWorkerProvider;
import alluxio.master.scheduler.JournaledJobMetaStore;
import alluxio.master.scheduler.Scheduler;
import alluxio.proto.journal.Job;
import alluxio.resource.CloseableResource;
import alluxio.scheduler.job.JobState;
import alluxio.security.authentication.AuthenticatedClientUser;
import alluxio.wire.FileInfo;
import alluxio.wire.WorkerInfo;
import alluxio.wire.WorkerNetAddress;

import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.Status;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

public final class SchedulerTest {

  @BeforeClass
  public static void before() {
    AuthenticatedClientUser.set("user");
  }

  @AfterClass
  public static void after() {
    AuthenticatedClientUser.remove();
  }

  @Test
  public void testGetActiveWorkers() throws IOException {
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    CloseableResource<BlockWorkerClient> blockWorkerClient = mock(CloseableResource.class);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    when(fsMaster.getWorkerInfoList())
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(1).setAddress(
                new WorkerNetAddress().setHost("worker1").setRpcPort(1234)),
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234))))
        .thenThrow(new UnavailableException("test"))
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234))))
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(1).setAddress(
                new WorkerNetAddress().setHost("worker1").setRpcPort(1234)),
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234))));
    when(fileSystemContext.acquireBlockWorkerClient(any())).thenReturn(blockWorkerClient);
    assertEquals(0, scheduler
        .getActiveWorkers().size());
    scheduler.updateWorkers();
    assertEquals(2, scheduler
        .getActiveWorkers().size());
    scheduler.updateWorkers();
    assertEquals(2, scheduler
        .getActiveWorkers().size());
    scheduler.updateWorkers();
    assertEquals(1, scheduler
        .getActiveWorkers().size());
    scheduler.updateWorkers();
    assertEquals(2, scheduler
        .getActiveWorkers().size());
  }

  @Test
  public void testSubmit() throws Exception {
    String validLoadPath = "/path/to/load";
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    FileIterable files =
        new FileIterable(fsMaster, validLoadPath, Optional.of("user"), false,
            LoadJob.QUALIFIED_FILE_FILTER);
    LoadJob loadJob =
        new LoadJob(validLoadPath, Optional.of("user"), "1", OptionalLong.empty(), false, true,
            files);
    assertTrue(scheduler.submitJob(loadJob));
    verify(journalContext).append(argThat(journalEntry -> journalEntry.hasLoadJob()
        && journalEntry.getLoadJob().getLoadPath().equals(validLoadPath)
        && journalEntry.getLoadJob().getState() == Job.PJobState.CREATED
        && !journalEntry.getLoadJob().hasBandwidth()
        && journalEntry.getLoadJob().getVerify()));
    assertEquals(1, scheduler
        .getJobs().size());
    LoadJob job = (LoadJob) scheduler.getJobs().get(loadJob.getDescription());
    assertEquals(OptionalLong.empty(), job.getBandwidth());
    assertTrue(job.isVerificationEnabled());
    loadJob =
        new LoadJob(validLoadPath, Optional.of("user"), "1", OptionalLong.of(1000), true, false,
            files);
    assertFalse(scheduler.submitJob(loadJob));
    verify(journalContext).append(argThat(journalEntry -> journalEntry.hasLoadJob()
        && journalEntry.getLoadJob().getLoadPath().equals(validLoadPath)
        && journalEntry.getLoadJob().getState() == Job.PJobState.CREATED
        && journalEntry.getLoadJob().getBandwidth() == 1000
        && !journalEntry.getLoadJob().getPartialListing()  // we don't update partialListing
        && !journalEntry.getLoadJob().getVerify()));
    assertEquals(1, scheduler
        .getJobs().size());
    job = (LoadJob) scheduler.getJobs().get(loadJob.getDescription());
    assertEquals(1000, job.getBandwidth().getAsLong());
    assertFalse(job.isVerificationEnabled());
  }

  @Test
  public void testStop() throws Exception {
    String validLoadPath = "/path/to/load";
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    FileIterable files =
        new FileIterable(fsMaster, validLoadPath, Optional.of("user"), false,
            LoadJob.QUALIFIED_FILE_FILTER);
    LoadJob job =
        new LoadJob(validLoadPath, Optional.of("user"), "1", OptionalLong.of(100), false, true,
            files);
    assertTrue(scheduler.submitJob(job));
    verify(journalContext, times(1)).append(any());
    verify(journalContext).append(argThat(journalEntry -> journalEntry.hasLoadJob()
        && journalEntry.getLoadJob().getLoadPath().equals(validLoadPath)
        && journalEntry.getLoadJob().getState() == Job.PJobState.CREATED
        && journalEntry.getLoadJob().getBandwidth() == 100
        && journalEntry.getLoadJob().getVerify()));
    assertTrue(scheduler.stopJob(job.getDescription()));
    verify(journalContext, times(2)).append(any());
    verify(journalContext).append(argThat(journalEntry -> journalEntry.hasLoadJob()
        && journalEntry.getLoadJob().getLoadPath().equals(validLoadPath)
        && journalEntry.getLoadJob().getState() == Job.PJobState.STOPPED
        && journalEntry.getLoadJob().getBandwidth() == 100
        && journalEntry.getLoadJob().getVerify()
        && journalEntry.getLoadJob().hasEndTime()));
    assertFalse(scheduler.stopJob(job.getDescription()));
    verify(journalContext, times(2)).append(any());
    assertFalse(scheduler.stopJob(JobDescription.newBuilder().setPath("/does/not/exist").build()));
    verify(journalContext, times(2)).append(any());
    assertFalse(scheduler.submitJob(job));
    verify(journalContext, times(3)).append(any());
    assertTrue(scheduler.stopJob(job.getDescription()));
    verify(journalContext, times(4)).append(any());
  }

  @Test
  public void testSubmitExceedsCapacity() throws Exception {
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    IntStream.range(0, 100).forEach(
        i -> {
          String path = String.format("/path/to/load/%d", i);
          FileIterable files = new FileIterable(fsMaster, path, Optional.of("user"), false,
              LoadJob.QUALIFIED_FILE_FILTER);
          assertTrue(scheduler.submitJob(
              new LoadJob(path, Optional.of("user"), "1", OptionalLong.empty(), false, true,
                  files)));
        });
    FileIterable files =
        new FileIterable(fsMaster, "/path/to/load/101", Optional.of("user"), false,
            LoadJob.QUALIFIED_FILE_FILTER);
    assertThrows(ResourceExhaustedRuntimeException.class, () -> scheduler.submitJob(
        new LoadJob("/path/to/load/101", Optional.of("user"), "1", OptionalLong.empty(), false,
            true, files)));
  }

  @Test
  public void testScheduling() throws Exception {
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    CloseableResource<BlockWorkerClient> blockWorkerClientResource = mock(CloseableResource.class);
    BlockWorkerClient blockWorkerClient = mock(BlockWorkerClient.class);
    when(fsMaster.getWorkerInfoList())
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(1).setAddress(
                new WorkerNetAddress().setHost("worker1").setRpcPort(1234)),
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234))))
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234))))
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(1).setAddress(
                new WorkerNetAddress().setHost("worker1").setRpcPort(1234)),
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234)),
            new WorkerInfo().setId(3).setAddress(
                new WorkerNetAddress().setHost("worker3").setRpcPort(1234)),
            new WorkerInfo().setId(4).setAddress(
                new WorkerNetAddress().setHost("worker4").setRpcPort(1234)),
            new WorkerInfo().setId(5).setAddress(
                new WorkerNetAddress().setHost("worker5").setRpcPort(1234)),
            new WorkerInfo().setId(6).setAddress(
                new WorkerNetAddress().setHost("worker6").setRpcPort(1234)),
            new WorkerInfo().setId(7).setAddress(
                new WorkerNetAddress().setHost("worker7").setRpcPort(1234)),
            new WorkerInfo().setId(8).setAddress(
                new WorkerNetAddress().setHost("worker8").setRpcPort(1234)),
            new WorkerInfo().setId(9).setAddress(
                new WorkerNetAddress().setHost("worker9").setRpcPort(1234)),
            new WorkerInfo().setId(10).setAddress(
                new WorkerNetAddress().setHost("worker10").setRpcPort(1234))));
    List<FileInfo> fileInfos = generateRandomFileInfo(100, 50, 64 * Constants.MB);
    when(fsMaster.listStatus(any(), any()))
        .thenReturn(fileInfos)
        .thenReturn(fileWithBlockLocations(fileInfos, 0.95))
        .thenReturn(fileWithBlockLocations(fileInfos, 1.1));
    when(fileSystemContext.acquireBlockWorkerClient(any())).thenReturn(blockWorkerClientResource);
    when(blockWorkerClientResource.get()).thenReturn(blockWorkerClient);
    AtomicInteger iteration = new AtomicInteger();
    when(blockWorkerClient.load(any())).thenAnswer(invocation -> {
      LoadRequest request = invocation.getArgument(0);
      return buildResponseFuture(request, iteration);
    });
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    String path = "test";
    FileIterable files = new FileIterable(fsMaster, path, Optional.of("user"), false,
        LoadJob.QUALIFIED_FILE_FILTER);
    LoadJob loadJob = new LoadJob(path, Optional.of("user"), "1",
        OptionalLong.of(1000), false, true, files);
    scheduler.submitJob(loadJob);
    verify(journalContext).append(argThat(journalEntry -> journalEntry.hasLoadJob()
        && journalEntry.getLoadJob().getLoadPath().equals(path)
        && journalEntry.getLoadJob().getState() == Job.PJobState.CREATED
        && journalEntry.getLoadJob().getBandwidth() == 1000
        && journalEntry.getLoadJob().getVerify()));

    scheduler.start();
    while (!scheduler
        .getJobProgress(loadJob.getDescription(), JobProgressReportFormat.TEXT, false)
        .contains("SUCCEEDED")) {
      assertFalse(scheduler.submitJob(
          new LoadJob(path, Optional.of("user"), "1", OptionalLong.of(1000), false, true, files)));
      Thread.sleep(1000);
    }
    Thread.sleep(1000);
    scheduler.stop();
    assertEquals(JobState.SUCCEEDED, loadJob.getJobState());
    assertEquals(0, loadJob.getCurrentBlockCount());
    verify(journalContext).append(argThat(journalEntry -> journalEntry.hasLoadJob()
        && journalEntry.getLoadJob().getLoadPath().equals(path)
        && journalEntry.getLoadJob().getState() == Job.PJobState.SUCCEEDED
        && journalEntry.getLoadJob().getBandwidth() == 1000
        && journalEntry.getLoadJob().getVerify()));
    assertTrue(scheduler.submitJob(new LoadJob(path, "user", OptionalLong.of(1000), files)));
  }

  private ListenableFuture<LoadResponse> buildResponseFuture(LoadRequest request,
      AtomicInteger iteration) {
    int failureRequestIteration = 50;
    int exceptionRequestIteration = 70;

    iteration.getAndIncrement();
    List<BlockStatus> status;
    if (iteration.get() == exceptionRequestIteration) {
      // Test worker exception
      SettableFuture<LoadResponse> responseFuture = SettableFuture.create();
      responseFuture.setException(new TimeoutException());
      return responseFuture;
    }
    else if (iteration.get() == failureRequestIteration) {
      // Test worker failing the whole request
      status = generateRandomBlockStatus(request.getBlocksList(), 1);
    }
    else {
      status = generateRandomBlockStatus(request.getBlocksList(), 0.01);
    }
    LoadResponse.Builder response = LoadResponse.newBuilder();
    if (status.stream().allMatch(s -> s.getCode() == Status.OK.getCode().value())) {
      response.setStatus(TaskStatus.SUCCESS);
    }
    else if (status.stream().allMatch(s -> s.getCode() != Status.OK.getCode().value())) {
      response.setStatus(TaskStatus.FAILURE)
          .addAllBlockStatus(status);
    }
    else {
      response.setStatus(TaskStatus.PARTIAL_FAILURE)
          .addAllBlockStatus(status.stream()
              .filter(s -> s.getCode() != Status.OK.getCode().value())
              .collect(ImmutableList.toImmutableList()));
    }
    SettableFuture<LoadResponse> responseFuture = SettableFuture.create();
    responseFuture.set(response.build());
    return responseFuture;
  }

  @Test
  public void testSchedulingFullCapacity() throws Exception {
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    CloseableResource<BlockWorkerClient> blockWorkerClientResource = mock(CloseableResource.class);
    BlockWorkerClient blockWorkerClient = mock(BlockWorkerClient.class);
    ImmutableList.Builder<WorkerInfo> workerInfos = ImmutableList.builder();
    for (int i = 0; i < 1000; i++) {
      workerInfos.add(new WorkerInfo().setId(i).setAddress(
          new WorkerNetAddress().setHost("worker" + i).setRpcPort(1234)));
    }
    when(fsMaster.getWorkerInfoList())
        .thenReturn(workerInfos.build());
    List<FileInfo> fileInfos = generateRandomFileInfo(2000, 50, 64 * Constants.MB);
    when(fsMaster.listStatus(any(), any()))
        .thenReturn(fileInfos);

    when(fileSystemContext.acquireBlockWorkerClient(any())).thenReturn(blockWorkerClientResource);
    when(blockWorkerClientResource.get()).thenReturn(blockWorkerClient);
    when(blockWorkerClient.load(any())).thenAnswer(invocation -> {
      LoadResponse.Builder response = LoadResponse.newBuilder().setStatus(TaskStatus.SUCCESS);
      SettableFuture<LoadResponse> responseFuture = SettableFuture.create();
      responseFuture.set(response.build());
      return responseFuture;
    });
    FileIterable files =
        new FileIterable(fsMaster, "test", Optional.of("user"), false,
            LoadJob.QUALIFIED_FILE_FILTER);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    for (int i = 0; i < 100; i++) {
      LoadJob loadJob = new LoadJob("test" + i, "user", OptionalLong.of(1000), files);
      scheduler.submitJob(loadJob);
    }
    assertThrows(ResourceExhaustedRuntimeException.class, () -> scheduler.submitJob(
        new LoadJob("/way/too/many", "user", OptionalLong.empty(), files)));
    scheduler.start();
    while (scheduler
        .getJobs().values().stream()
        .anyMatch(loadJob -> loadJob.getJobState() != JobState.SUCCEEDED)) {
      Thread.sleep(1000);
    }
    scheduler.stop();
  }

  @Test
  public void testSchedulingWithException() throws Exception {
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    CloseableResource<BlockWorkerClient> blockWorkerClientResource = mock(CloseableResource.class);
    BlockWorkerClient blockWorkerClient = mock(BlockWorkerClient.class);
    when(fsMaster.getWorkerInfoList())
        .thenReturn(ImmutableList.of(
            new WorkerInfo().setId(1).setAddress(
                new WorkerNetAddress().setHost("worker1").setRpcPort(1234)),
            new WorkerInfo().setId(2).setAddress(
                new WorkerNetAddress().setHost("worker2").setRpcPort(1234))));
    when(fileSystemContext.acquireBlockWorkerClient(any())).thenReturn(blockWorkerClientResource);
    when(blockWorkerClientResource.get()).thenReturn(blockWorkerClient);
    List<FileInfo> fileInfos = generateRandomFileInfo(100, 10, 64 * Constants.MB);
    when(fsMaster.listStatus(any(), any()))
        // Non-retryable exception, first load job should fail
        .thenThrow(AccessControlException.class)
        // Retryable exception, second load job should succeed
        .thenThrow(new ResourceExhaustedRuntimeException("test", true))
        .thenReturn(fileInfos);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    scheduler.start();
    FileIterable files =
        new FileIterable(fsMaster, "test", Optional.of("user"), false,
            LoadJob.QUALIFIED_FILE_FILTER);
    LoadJob job = new LoadJob("test", "user", OptionalLong.of(1000), files);
    scheduler.submitJob(job);
    while (!scheduler
        .getJobProgress(job.getDescription(), JobProgressReportFormat.TEXT, false)
        .contains("FAILED")) {
      Thread.sleep(1000);
    }
    when(blockWorkerClient.load(any())).thenAnswer(invocation -> {
      LoadResponse.Builder response = LoadResponse.newBuilder().setStatus(TaskStatus.SUCCESS);
      SettableFuture<LoadResponse> responseFuture = SettableFuture.create();
      responseFuture.set(response.build());
      return responseFuture;
    });
    job = new LoadJob("test", "user", OptionalLong.of(1000), files);
    scheduler.submitJob(job);
    while (!scheduler
        .getJobProgress(job.getDescription(), JobProgressReportFormat.TEXT, false)
        .contains("SUCCEEDED")) {
      Thread.sleep(1000);
    }
    scheduler.stop();
  }

  @Test
  public void testJobRetention() throws Exception {
    Configuration.modifiableGlobal().set(PropertyKey.JOB_RETENTION_TIME, "0ms", Source.RUNTIME);
    FileSystemMaster fsMaster = mock(FileSystemMaster.class);
    FileSystemContext fileSystemContext = mock(FileSystemContext.class);
    JournalContext journalContext = mock(JournalContext.class);
    when(fsMaster.createJournalContext()).thenReturn(journalContext);
    DefaultWorkerProvider workerProvider =
        new DefaultWorkerProvider(fsMaster, fileSystemContext);
    Scheduler scheduler = new Scheduler(workerProvider, new JournaledJobMetaStore(fsMaster));
    scheduler.start();
    IntStream
        .range(0, 5)
        .forEach(i -> {
          String path = String.format("/load/%d", i);
          FileIterable files = new FileIterable(fsMaster, path, Optional.of("user"),
              false, LoadJob.QUALIFIED_FILE_FILTER);
          assertTrue(scheduler.submitJob(
              new LoadJob(path, Optional.of("user"), "1",
                  OptionalLong.empty(), false, true, files)));
        });
    assertEquals(5, scheduler
        .getJobs().size());
    scheduler
        .getJobs()
        .get(JobDescription
            .newBuilder()
            .setPath("/load/1")
            .setType("load")
            .build())
        .setJobState(JobState.VERIFYING);
    scheduler
        .getJobs()
        .get(JobDescription
            .newBuilder()
            .setPath("/load/2")
            .setType("load")
            .build())
        .setJobState(JobState.FAILED);
    scheduler
        .getJobs()
        .get(JobDescription
            .newBuilder()
            .setPath("/load/3")
            .setType("load")
            .build())
        .setJobState(JobState.SUCCEEDED);
    scheduler
        .getJobs()
        .get(JobDescription
            .newBuilder()
            .setPath("/load/4")
            .setType("load")
            .build())
        .setJobState(JobState.STOPPED);
    scheduler.cleanupStaleJob();
    assertEquals(2, scheduler
        .getJobs().size());
    assertTrue(scheduler
        .getJobs().containsKey(JobDescription
            .newBuilder()
            .setPath("/load/0")
            .setType("load")
            .build()));
    assertTrue(scheduler
        .getJobs().containsKey(JobDescription
            .newBuilder()
            .setPath("/load/1")
            .setType("load")
            .build()));
    IntStream.range(2, 5).forEach(
        i -> assertFalse(scheduler
            .getJobs().containsKey(JobDescription
            .newBuilder()
            .setPath("/load/" + i)
            .setType("load")
            .build())));
    Configuration.modifiableGlobal().unset(PropertyKey.JOB_RETENTION_TIME);
  }
}
