code: @Override
  	public CompletableFuture<?> releaseSlot(@Nullable Throwable cause) {
  		if (!isCanceled()) {
  			final CompletableFuture<?> terminationFuture;
  
  			if (payload != null) {
  				// trigger the failure of the slot payload
  				payload.fail(cause != null ? cause : new FlinkException("TaskManager was lost/killed: " + getTaskManagerLocation()));
  
  				// wait for the termination of the payload before releasing the slot
  				terminationFuture = payload.getTerminalStateFuture();
  			} else {
  				terminationFuture = CompletableFuture.completedFuture(null);
  			}
  
  			terminationFuture.whenComplete(
  				(Object ignored, Throwable throwable) -> {
  					// release directly (if we are directly allocated),
  					// otherwise release through the parent shared slot
  					if (getParent() == null) {
  						// we have to give back the slot to the owning instance
  						if (markCancelled()) {
  							try {
  								getOwner().returnLogicalSlot(this);
  								releaseFuture.complete(null);
  							} catch (Exception e) {
  								releaseFuture.completeExceptionally(e);
  							}
  						}
  					} else {
  						// we have to ask our parent to dispose us
  						getParent().releaseChild(this);
  
  						releaseFuture.complete(null);
  					}
  				});
  		}
  
  		return releaseFuture;
  	}
docs: ------------------------------------------------------------------------