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());
  }
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 ( ) ) ; }
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