### Question:
SqlTaskManager implements TaskManager, Closeable { public void removeOldTasks() { DateTime oldestAllowedTask = DateTime.now().minus(infoCacheTime.toMillis()); tasks.asMap().values().stream() .map(SqlTask::getTaskInfo) .filter(Objects::nonNull) .forEach(taskInfo -> { TaskId taskId = taskInfo.getTaskStatus().getTaskId(); try { DateTime endTime = taskInfo.getStats().getEndTime(); if (endTime != null && endTime.isBefore(oldestAllowedTask)) { tasks.asMap().remove(taskId); } } catch (RuntimeException e) { log.warn(e, "Error while inspecting age of complete task %s", taskId); } }); } @Inject  SqlTaskManager(
            LocalExecutionPlanner planner,
            LocationFactory locationFactory,
            TaskExecutor taskExecutor,
            SplitMonitor splitMonitor,
            NodeInfo nodeInfo,
            LocalMemoryManager localMemoryManager,
            TaskManagementExecutor taskManagementExecutor,
            TaskManagerConfig config,
            NodeMemoryConfig nodeMemoryConfig,
            LocalSpillManager localSpillManager,
            NodeSpillConfig nodeSpillConfig,
            GcMonitor gcMonitor); @Override synchronized void updateMemoryPoolAssignments(MemoryPoolAssignmentsRequest assignments); @PostConstruct void start(); @Override @PreDestroy void close(); @Managed @Flatten SqlTaskIoStats getIoStats(); @Managed(description = "Task notification executor") @Nested ThreadPoolExecutorMBean getTaskNotificationExecutor(); @Managed(description = "Failed tasks counter") @Nested CounterStat getFailedTasks(); List<SqlTask> getAllTasks(); @Override List<TaskInfo> getAllTaskInfo(); @Override TaskInfo getTaskInfo(TaskId taskId); @Override TaskStatus getTaskStatus(TaskId taskId); @Override ListenableFuture<TaskInfo> getTaskInfo(TaskId taskId, long currentVersion); @Override String getTaskInstanceId(TaskId taskId); @Override ListenableFuture<TaskStatus> getTaskStatus(TaskId taskId, long currentVersion); @Override VersionedDynamicFilterDomains acknowledgeAndGetNewDynamicFilterDomains(TaskId taskId, long currentDynamicFiltersVersion); @Override TaskInfo updateTask(Session session, TaskId taskId, Optional<PlanFragment> fragment, List<TaskSource> sources, OutputBuffers outputBuffers, OptionalInt totalPartitions); @Override ListenableFuture<BufferResult> getTaskResults(TaskId taskId, OutputBufferId bufferId, long startingSequenceId, DataSize maxSize); @Override void acknowledgeTaskResults(TaskId taskId, OutputBufferId bufferId, long sequenceId); @Override TaskInfo abortTaskResults(TaskId taskId, OutputBufferId bufferId); @Override TaskInfo cancelTask(TaskId taskId); @Override TaskInfo abortTask(TaskId taskId); void removeOldTasks(); void failAbandonedTasks(); @Override void addStateChangeListener(TaskId taskId, StateChangeListener<TaskState> stateChangeListener); @VisibleForTesting QueryContext getQueryContext(QueryId queryId);  }### Answer:
@Test public void testRemoveOldTasks() throws Exception { try (SqlTaskManager sqlTaskManager = createSqlTaskManager(new TaskManagerConfig().setInfoMaxAge(new Duration(5, TimeUnit.MILLISECONDS)))) { TaskId taskId = TASK_ID; TaskInfo taskInfo = createTask(sqlTaskManager, taskId, createInitialEmptyOutputBuffers(PARTITIONED).withBuffer(OUT, 0).withNoMoreBufferIds()); assertEquals(taskInfo.getTaskStatus().getState(), TaskState.RUNNING); taskInfo = sqlTaskManager.cancelTask(taskId); assertEquals(taskInfo.getTaskStatus().getState(), TaskState.CANCELED); taskInfo = sqlTaskManager.getTaskInfo(taskId); assertEquals(taskInfo.getTaskStatus().getState(), TaskState.CANCELED); Thread.sleep(100); sqlTaskManager.removeOldTasks(); for (TaskInfo info : sqlTaskManager.getAllTaskInfo()) { assertNotEquals(info.getTaskStatus().getTaskId(), taskId); } } }