code: private void removeWorker(final Worker worker)
    {
      log.info("Kaboom! Worker[%s] removed!", worker.getHost());
  
      final ZkWorker zkWorker = zkWorkers.get(worker.getHost());
      if (zkWorker != null) {
        try {
          scheduleTasksCleanupForWorker(worker.getHost(), getAssignedTasks(worker));
        }
        catch (Exception e) {
          throw new RuntimeException(e);
        }
        finally {
          try {
            zkWorker.close();
          }
          catch (Exception e) {
            log.error(e, "Exception closing worker[%s]!", worker.getHost());
          }
          zkWorkers.remove(worker.getHost());
          checkBlackListedNodes();
        }
      }
      lazyWorkers.remove(worker.getHost());
    }
docs: When a ephemeral worker node disappears from ZK, incomplete running tasks will be retried by
  the logic in the status listener. We still have to make sure there are no tasks assigned
  to the worker but not yet running.
  
  @param worker - the removed worker