### Question:
ProvisioningService extends AbstractIdleService { @VisibleForTesting void resumeTasks(Consumer<ProgramRunId> taskCleanup) throws IOException { List<ProvisioningTaskInfo> clusterTaskInfos = getInProgressTasks(); for (ProvisioningTaskInfo provisioningTaskInfo : clusterTaskInfos) { State serviceState = state(); if (serviceState != State.STARTING && serviceState != State.RUNNING) { return; } ProvisioningTaskKey taskKey = provisioningTaskInfo.getTaskKey(); Optional<Future<Void>> taskFuture = taskExecutor.getFuture(taskKey); if (taskFuture.isPresent()) { continue; } ProgramRunId programRunId = provisioningTaskInfo.getProgramRunId(); ProvisioningOp provisioningOp = provisioningTaskInfo.getProvisioningOp(); String provisionerName = provisioningTaskInfo.getProvisionerName(); Provisioner provisioner = provisionerInfo.get().provisioners.get(provisionerName); if (provisioner == null) { LOG.error("Could not provision cluster for program run {} because provisioner {} no longer exists.", programRunId, provisionerName); provisionerNotifier.orphaned(programRunId); provisionerStore.deleteTaskInfo(provisioningTaskInfo.getProgramRunId()); continue; } Runnable task; switch (provisioningOp.getType()) { case PROVISION: task = createProvisionTask(provisioningTaskInfo, provisioner); break; case DEPROVISION: task = createDeprovisionTask(provisioningTaskInfo, provisioner, taskCleanup); break; default: LOG.error("Skipping unknown provisioning task type {}", provisioningOp.getType()); continue; } if (provisioningOp.getStatus() != ProvisioningOp.Status.CANCELLED && provisioningOp.getStatus() != ProvisioningOp.Status.CREATED) { LOG.info("Resuming provisioning task for run {} of type {} in state {}.", provisioningTaskInfo.getProgramRunId(), provisioningTaskInfo.getProvisioningOp().getType(), provisioningTaskInfo.getProvisioningOp().getStatus()); } task.run(); } } @Inject  ProvisioningService(CConfiguration cConf, ProvisionerProvider provisionerProvider,
                      ProvisionerConfigProvider provisionerConfigProvider,
                      ProvisionerNotifier provisionerNotifier, LocationFactory locationFactory,
                      SecureStore secureStore, ProgramStateWriter programStateWriter,
                      ProvisionerStore provisionerStore, TransactionRunner transactionRunner,
                      MetricsCollectionService metricsCollectionService); ClusterStatus getClusterStatus(ProgramRunId programRunId, ProgramOptions programOptions,
                                        Cluster cluster, String userId); Optional<ProvisioningTaskInfo> cancelProvisionTask(ProgramRunId programRunId); Runnable provision(ProvisionRequest provisionRequest, StructuredTableContext context); Optional<RuntimeJobManager> getRuntimeJobManager(ProgramRunId programRunId, ProgramOptions programOptions); Runnable deprovision(ProgramRunId programRunId, StructuredTableContext context); Collection<ProvisionerDetail> getProvisionerDetails(); @Nullable ProvisionerDetail getProvisionerDetail(String name); void validateProperties(String provisionerName, Map<String, String> properties);  }### Answer:
@Test public void testScanForTasks() throws Exception { TaskFields taskFields = createTaskInfo(new MockProvisioner.PropertyBuilder().build()); ProvisioningOp op = new ProvisioningOp(ProvisioningOp.Type.PROVISION, ProvisioningOp.Status.POLLING_CREATE); Cluster cluster = new Cluster("name", ClusterStatus.CREATING, Collections.emptyList(), Collections.emptyMap()); ProvisioningTaskInfo taskInfo = new ProvisioningTaskInfo(taskFields.programRunId, taskFields.programDescriptor, taskFields.programOptions, Collections.emptyMap(), MockProvisioner.NAME, "Bob", op, Locations.toLocation(TEMP_FOLDER.newFolder()).toURI(), cluster); provisionerStore.putTaskInfo(taskInfo); provisioningService.resumeTasks(t -> { }); ProvisioningTaskKey taskKey = new ProvisioningTaskKey(taskFields.programRunId, ProvisioningOp.Type.PROVISION); waitForExpectedProvisioningState(taskKey, ProvisioningOp.Status.CREATED); }