code: Optional<Collection<Tasklet>> removeWorker(final String id) {
      lock.lock();
      try {
        if (!terminated) {
          final VortexWorkerManager vortexWorkerManager = this.runningWorkers.remove(id);
          if (vortexWorkerManager != null) {
            this.schedulingPolicy.workerRemoved(vortexWorkerManager);
            return Optional.ofNullable(vortexWorkerManager.removed());
          } else {
            // Called before addWorker (e.g. RM preempted the resource before the Evaluator started)
            removedBeforeAddedWorkers.add(id);
            return Optional.empty();
          }
        } else {
          // No need to return anything since it is terminated
          return Optional.empty();
        }
      } finally {
        try {
          workerAggregateFunctionMap.remove(id);
        } finally {
          lock.unlock();
        }
      }
    }
docs: Concurrency: Called by multiple threads.
  Parameter: Called exactly once per id.