code: protected void closeTaskManagerConnection(final ResourceID resourceID, final Exception cause) {
  		taskManagerHeartbeatManager.unmonitorTarget(resourceID);
  
  		WorkerRegistration<WorkerType> workerRegistration = taskExecutors.remove(resourceID);
  
  		if (workerRegistration != null) {
  			log.info("Closing TaskExecutor connection {} because: {}", resourceID, cause.getMessage());
  
  			// TODO :: suggest failed task executor to stop itself
  			slotManager.unregisterTaskManager(workerRegistration.getInstanceID());
  
  			workerRegistration.getTaskExecutorGateway().disconnectResourceManager(cause);
  		} else {
  			log.debug(
  				"No open TaskExecutor connection {}. Ignoring close TaskExecutor connection. Closing reason was: {}",
  				resourceID,
  				cause.getMessage());
  		}
  	}
docs: This method should be called by the framework once it detects that a currently registered
  task executor has failed.
  
  @param resourceID Id of the TaskManager that has failed.
  @param cause The exception which cause the TaskManager failed.