docs: private void manage() throws InterruptedException
    {
      log.info("Beginning management in %s.", config.getStartDelay());
      Thread.sleep(config.getStartDelay().getMillis());
  
      // Ignore return value- we'll get the IDs and futures from getKnownTasks later.
      taskRunner.restore();
  
      while (active) {
        giant.lock();
  
        try {
          // Task futures available from the taskRunner
          final Map<String, ListenableFuture<TaskStatus>> runnerTaskFutures = new HashMap<>();
          for (final TaskRunnerWorkItem workItem : taskRunner.getKnownTasks()) {
            runnerTaskFutures.put(workItem.getTaskId(), workItem.getResult());
          }
          // Attain futures for all active tasks (assuming they are ready to run).
          // Copy tasks list, as notifyStatus may modify it.
          for (final Task task : ImmutableList.copyOf(tasks)) {
            if (!taskFutures.containsKey(task.getId())) {
              final ListenableFuture<TaskStatus> runnerTaskFuture;
              if (runnerTaskFutures.containsKey(task.getId())) {
                runnerTaskFuture = runnerTaskFutures.get(task.getId());
              } else {
                // Task should be running, so run it.
                final boolean taskIsReady;
                try {
                  taskIsReady = task.isReady(taskActionClientFactory.create(task));
                }
                catch (Exception e) {
                  log.warn(e, "Exception thrown during isReady for task: %s", task.getId());
                  notifyStatus(task, TaskStatus.failure(task.getId()), "failed because of exception[%s]", e.getClass());
                  continue;
                }
                if (taskIsReady) {
                  log.info("Asking taskRunner to run: %s", task.getId());
                  runnerTaskFuture = taskRunner.run(task);
                } else {
                  continue;
                }
              }
              taskFutures.put(task.getId(), attachCallbacks(task, runnerTaskFuture));
            } else if (isTaskPending(task)) {
              // if the taskFutures contain this task and this task is pending, also let the taskRunner
              // to run it to guarantee it will be assigned to run
              // see https://github.com/apache/incubator-druid/pull/6991
              taskRunner.run(task);
            }
          }
          // Kill tasks that shouldn't be running
          final Set<String> tasksToKill = Sets.difference(
              runnerTaskFutures.keySet(),
              ImmutableSet.copyOf(
                  Lists.transform(
                      tasks,
                      new Function<Task, Object>()
                      {
                        @Override
                        public String apply(Task task)
                        {
                          return task.getId();
                        }
                      }
                  )
              )
          );
          if (!tasksToKill.isEmpty()) {
            log.info("Asking taskRunner to clean up %,d tasks.", tasksToKill.size());
            for (final String taskId : tasksToKill) {
              try {
                taskRunner.shutdown(
                    taskId,
                    "task is not in runnerTaskFutures[%s]",
                    runnerTaskFutures.keySet()
                );
              }
              catch (Exception e) {
                log.warn(e, "TaskRunner failed to clean up task: %s", taskId);
              }
            }
          }
          // awaitNanos because management may become necessary without this condition signalling,
          // due to e.g. tasks becoming ready when other folks mess with the TaskLockbox.
          managementMayBeNecessary.awaitNanos(MANAGEMENT_WAIT_TIMEOUT_NANOS);
        }
        finally {
          giant.unlock();
        }
      }
    }