docs: @Override
  	public CompletableFuture<Void> onStop() {
  		log.info("Stopping TaskExecutor {}.", getAddress());
  
  		Throwable throwable = null;
  
  		if (resourceManagerConnection != null) {
  			resourceManagerConnection.close();
  		}
  
  		for (JobManagerConnection jobManagerConnection : jobManagerConnections.values()) {
  			try {
  				disassociateFromJobManager(jobManagerConnection, new FlinkException("The TaskExecutor is shutting down."));
  			} catch (Throwable t) {
  				throwable = ExceptionUtils.firstOrSuppressed(t, throwable);
  			}
  		}
  
  		jobManagerHeartbeatManager.stop();
  
  		resourceManagerHeartbeatManager.stop();
  
  		try {
  			stopTaskExecutorServices();
  		} catch (Exception e) {
  			throwable = ExceptionUtils.firstOrSuppressed(e, throwable);
  		}
  
  		if (throwable != null) {
  			return FutureUtils.completedExceptionally(new FlinkException("Error while shutting the TaskExecutor down.", throwable));
  		} else {
  			log.info("Stopped TaskExecutor {}.", getAddress());
  			return CompletableFuture.completedFuture(null);
  		}
  	}