### Question:
RunRecordCorrectorService extends AbstractIdleService { @Override protected void startUp() throws Exception { LOG.info("Starting RunRecordCorrectorService"); localDatasetDeleterService = Executors .newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("local dataset deleter").build()); long interval = cConf.getLong(Constants.AppFabric.LOCAL_DATASET_DELETER_INTERVAL_SECONDS); if (interval <= 0) { LOG.warn("Invalid interval specified for the local dataset deleter {}. Setting it to 3600 seconds.", interval); interval = 3600L; } long initialDelay = cConf.getLong(Constants.AppFabric.LOCAL_DATASET_DELETER_INITIAL_DELAY_SECONDS); if (initialDelay <= 0) { LOG.warn("Invalid initial delay specified for the local dataset deleter {}. Setting it to 300 seconds.", initialDelay); initialDelay = 300L; } Runnable runnable = new LocalDatasetDeleterRunnable(namespaceAdmin, store, datasetFramework); localDatasetDeleterService.scheduleWithFixedDelay(runnable, initialDelay, interval, TimeUnit.SECONDS); } RunRecordCorrectorService(CConfiguration cConf, Store store, ProgramStateWriter programStateWriter,
                            ProgramRuntimeService runtimeService, NamespaceAdmin namespaceAdmin,
                            DatasetFramework datasetFramework); @VisibleForTesting  RunRecordCorrectorService(CConfiguration cConf, Store store, ProgramStateWriter programStateWriter,
                            ProgramRuntimeService runtimeService, NamespaceAdmin namespaceAdmin,
                            DatasetFramework datasetFramework,
                            long startTimeoutSecs, int txBatchSize);   }### Answer:
@Test public void testLocalDatasetDeleteion() throws Exception { deploy(WorkflowAppWithLocalDataset.class, 200, Constants.Gateway.API_VERSION_3_TOKEN, TEST_NAMESPACE1); final ProgramId workflow = new NamespaceId(TEST_NAMESPACE1) .app(WorkflowAppWithLocalDataset.APP_NAME) .workflow(WorkflowAppWithLocalDataset.WORKFLOW_NAME); startProgram(Id.Program.fromEntityId(workflow), ImmutableMap.of("dataset.*.keep.local", "true")); Tasks.waitFor(1, () -> getProgramRuns(Id.Program.fromEntityId(workflow), ProgramRunStatus.COMPLETED).size(), 5, TimeUnit.SECONDS); List<RunRecord> runRecords = getProgramRuns(Id.Program.fromEntityId(workflow), ProgramRunStatus.COMPLETED); Assert.assertEquals(1, runRecords.size()); String pid = runRecords.get(0).getPid(); final Map<String, String> properties = ImmutableMap.of(Constants.AppFabric.WORKFLOW_LOCAL_DATASET_PROPERTY, "true"); Collection<DatasetSpecificationSummary> instances = datasetFramework.getInstances(new NamespaceId(TEST_NAMESPACE1), properties); Assert.assertEquals(1, instances.size()); DatasetSpecificationSummary summary = instances.iterator().next(); Assert.assertTrue(summary.getName().endsWith(pid)); Map<String, String> updatedProperties = new HashMap<>(summary.getProperties()); updatedProperties.remove(Constants.AppFabric.WORKFLOW_KEEP_LOCAL); datasetFramework.updateInstance(new DatasetId(TEST_NAMESPACE1, summary.getName()), DatasetProperties.of(updatedProperties)); CConfiguration testConf = CConfiguration.create(); testConf.set(Constants.AppFabric.LOCAL_DATASET_DELETER_INTERVAL_SECONDS, "1"); testConf.set(Constants.AppFabric.LOCAL_DATASET_DELETER_INITIAL_DELAY_SECONDS, "1"); new LocalRunRecordCorrectorService(testConf, store, programStateWriter, runtimeService, namespaceAdmin, datasetFramework).startUp(); Tasks.waitFor(0, () -> datasetFramework.getInstances(new NamespaceId(TEST_NAMESPACE1), properties).size(), 30, TimeUnit.SECONDS, 1, TimeUnit.SECONDS); }