because both invocationRegistry.deregister() and future.complete() are idempotent.
195472
private void complete(Object value) {
        future.complete(value);
        if (context.invocationRegistry.deregister(this) && taskDoneCallback != null) {
            context.asyncExecutor.execute(taskDoneCallback);
        }
    }
137202
private CompletableFuture<ProcessorCollection> getProcessor(long streamSegmentId) {
        ProcessorCollection existingProcessor = this.processors.getOrDefault(streamSegmentId, null);
        if (existingProcessor != null) {
            if (closeIfNecessary(existingProcessor).isClosed()) {
                // Existing SegmentAggregator has become stale (most likely due to its SegmentMetadata being evicted),
                // so it has been closed and we need to create a new one.
                this.processors.remove(streamSegmentId);
            } else {
                return CompletableFuture.completedFuture(existingProcessor);
            }
        }

        // Get the SegmentAggregator's Metadata.
        UpdateableSegmentMetadata segmentMetadata = this.dataSource.getStreamSegmentMetadata(streamSegmentId);
        if (segmentMetadata == null) {
            return Futures.failedFuture(new DataCorruptionException(String.format(
                    "No StreamSegment with id '%d' is registered in the metadata.", streamSegmentId)));
        }

        // Then create the aggregator, and only register it after a successful initialization. Otherwise we risk
        // having a registered aggregator that is not initialized.
        SegmentAggregator newAggregator = new SegmentAggregator(segmentMetadata, this.dataSource, this.storage, this.config, this.timer, this.executor);
        ProcessorCollection pc = new ProcessorCollection(newAggregator, this.createProcessors.apply(segmentMetadata));
        try {
            CompletableFuture<Void> init = newAggregator.initialize(this.config.getFlushTimeout());
            Futures.exceptionListener(init, ex -> newAggregator.close());
            return init.thenApply(ignored -> {
                this.processors.put(streamSegmentId, pc);
                return pc;
            });
        } catch (Exception ex) {
            pc.close();
            throw ex;
        }
    }
474116
@Override
    public void deregisterAsynchConsumerCallback()
                    throws SISessionUnavailableException, SISessionDroppedException,
                    SIErrorException,
                    SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.entry(CoreSPIConsumerSession.tc, "deregisterAsynchConsumerCallback", this);
        //Deregister the AsynchConsumer by registering null with the LCP
        //the other params don't matter
        registerAsynchConsumerCallback(null, 1, 0, 1, null);
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConsumerSession.tc, "deregisterAsynchConsumerCallback");
    }
950631
private boolean checkImplementer(TypeElement providerImplementer, TypeElement providerType) {

    String verify = processingEnv.getOptions().get("verify");
    if (verify == null || !Boolean.valueOf(verify)) {
      return true;
    }

    // TODO: We're currently only enforcing the subtype relationship
    // constraint. It would be nice to enforce them all.

    Types types = processingEnv.getTypeUtils();

    return types.isSubtype(providerImplementer.asType(), providerType.asType());
  }
53790
@Trivial // traced by caller
    static <U> CompletionStage<U> completedStage(U value, Executor executor) {
        if (JAVA8) {
            return new ManagedCompletionStage<U>(CompletableFuture.completedFuture(value), executor, null);
        } else {
            ManagedCompletableFuture<U> stage = new ManagedCompletionStage<U>(executor);
            stage.super_complete(value);
            return stage;
        }
    }
327922
@Override
    @SuppressWarnings({ "PMD.EmptySynchronizedBlock", "PMD.EmptyCatchBlock",
        "PMD.AvoidCatchingThrowable" })
    public void run() {
        try {
            registerAsGenerator();
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    selector.select();
                    Set<SelectionKey> selected = selector.selectedKeys();
                    for (SelectionKey key : selected) {
                        ((NioHandler) key.attachment())
                            .handleOps(key.readyOps());
                    }
                    selected.clear();
                    synchronized (selectorGate) {
                        // Delay next iteration if another thread has the lock.
                        // "Find bugs" complains, but this is really okay.
                    }
                } catch (InterruptedIOException | InterruptedException
                        | Error e) {
                    break;
                } catch (Throwable e) {
                    // Ignore anything else, this loop is crucial.
                }
            }
        } finally {
            unregisterAsGenerator();
        }
    }
1274725
private void unregister(boolean retryOnFailure, CompletableFuture<Void> future) {
    long sessionId = state.getSessionId();

    // If the session is already closed, skip the unregister attempt.
    if (state.getState() == Session.State.CLOSED) {
      future.complete(null);
      return;
    }

    state.getLogger().debug("Unregistering session: {}", sessionId);

    // If a keep-alive request is already pending, cancel it.
    if (keepAlive != null) {
      keepAlive.cancel();
      keepAlive = null;
    }

    // If the current sessions state is unstable, reset the connection before sending an unregister request.
    if (state.getState() == Session.State.UNSTABLE) {
      connection.reset();
    }

    UnregisterRequest request = UnregisterRequest.builder()
      .withSession(sessionId)
      .build();

    state.getLogger().trace("{} - Sending {}", sessionId, request);
    connection.<UnregisterRequest, UnregisterResponse>sendAndReceive(request).whenComplete((response, error) -> {
      if (state.getState() != Session.State.CLOSED) {
        if (error == null) {
          state.getLogger().trace("{} - Received {}", sessionId, response);
          // If the request was successful, update the session state and complete the close future.
          if (response.status() == Response.Status.OK) {
            state.setState(Session.State.CLOSED);
            future.complete(null);
          }
          // If the session is unknown, immediate expire the session and complete the close future.
          else if (response.error() == CopycatError.Type.UNKNOWN_SESSION_ERROR) {
            state.setState(Session.State.EXPIRED);
            future.complete(null);
          }
          // If a leader is still set in the address selector, unset the leader and send another unregister attempt.
          // This will ensure that the address selector selects all servers without filtering on the leader.
          else if (retryOnFailure && connection.leader() != null) {
            connection.reset(null, connection.servers());
            unregister(false, future);
          }
          // If no leader was set, set the session state to unstable and fail the unregister attempt.
          else {
            state.setState(Session.State.UNSTABLE);
            future.completeExceptionally(new ClosedSessionException("failed to unregister session"));
          }
        }
        // If a leader is still set in the address selector, unset the leader and send another unregister attempt.
        // This will ensure that the address selector selects all servers without filtering on the leader.
        else if (retryOnFailure && connection.leader() != null) {
          connection.reset(null, connection.servers());
          unregister(false, future);
        }
        // If no leader was set, set the session state to unstable and schedule another unregister attempt.
        else {
          state.setState(Session.State.UNSTABLE);
          future.completeExceptionally(new ClosedSessionException("failed to unregister session"));
        }
      }
    });
  }
584480
private <R, U> CompletionStage<U> doApplyToEitherAsync(
            CompletionStage<? extends R> first,
            CompletionStage<? extends R> second,
            Function<? super R, U> fn,
            Executor executor) {
    	CompletableCompletionStage<R> nextStage = newCompletableCompletionStage();

        // only the first result is accepted by completion stage,
        // the other one is ignored
        BiConsumer<R, Throwable> action = completeHandler(nextStage);
        first.whenComplete(action);
        second.whenComplete(action);

        return nextStage.thenApplyAsync(fn, executor);
    }
1075013
@Override
    public void deregisterControlAdapterMBean()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "deregisterControlAdapterAsMBean");
        subscriptionControlAdaptor.dereferenceControllable();
        subscriptionControlAdaptor = null;
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "deregisterControlAdapterAsMBean");
    }
1362691
private static <T> void retryOperation(
			final CompletableFuture<T> resultFuture,
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Executor executor) {

		if (!resultFuture.isDone()) {
			final CompletableFuture<T> operationFuture = operation.get();

			operationFuture.whenCompleteAsync(
				(t, throwable) -> {
					if (throwable != null) {
						if (throwable instanceof CancellationException) {
							resultFuture.completeExceptionally(new RetryException("Operation future was cancelled.", throwable));
						} else {
							if (retries > 0) {
								retryOperation(
									resultFuture,
									operation,
									retries - 1,
									executor);
							} else {
								resultFuture.completeExceptionally(new RetryException("Could not complete the operation. Number of retries " +
									"has been exhausted.", throwable));
							}
						}
					} else {
						resultFuture.complete(t);
					}
				},
				executor);

			resultFuture.whenComplete(
				(t, throwable) -> operationFuture.cancel(false));
		}
	}
494768
@Override
    public void deregister(Controllable controllable, ControllableType type)
                    throws NotRegisteredException, SIBExceptionInvalidValue {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "deregister", new Object[] { controllable, type });
        try {
            if (type == ControllableType.QUEUE_POINT) {
                ServiceRegistration<QueueMBean> qMBean = (ServiceRegistration<QueueMBean>) this.serviceObjects.get(controllable.getName());
                qMBean.unregister();
            } else if (type == ControllableType.PUBLICATION_POINT) {
                ServiceRegistration<TopicMBean> topicMBean = (ServiceRegistration<TopicMBean>) this.serviceObjects.get(controllable.getName());
                topicMBean.unregister();
            } else if (type == ControllableType.SUBSCRIPTION_POINT) {
                ServiceRegistration<SubscriberMBean> subMBean = (ServiceRegistration<SubscriberMBean>) this.serviceObjects.get(controllable.getName());
                subMBean.unregister();
            }
        } catch (Exception e) {
            SibTr.exception(tc, e);
        }

        deregister(controllable);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "deregister");
        return;
    }
560150
private final void complete(boolean reUse,
                                Transaction transaction)
                    throws ObjectManagerException {
        final String methodName = "complete";
        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.entry(this, cclass, methodName, new Object[] { new Boolean(reUse),
                                                                transaction });

        // No longer any need to include this transaction in a Checkpoint.
        requiresPersistentCheckpoint = false;
        // Clear any remaining state.
        logicalUnitOfWork.XID = null;
        transactionLock = new TransactionLock(this);
        includedManagedObjects.clear();
        callbackTokens.clear();
        allPersistentTokensToNotify.clear();
        loggedSerializedBytes.clear();
        logSequenceNumbers.clear();
        managedObjectSequenceNumbers.clear();
        useCount++;

        if (reUse) { // Reset the transaction for further use.
            // Note that we do not clear the transactionReference because the caller is still holding it.
            // If the caller releases his reference then the reference will be found by ObjectManagerState
            // and this InternalTransaction may be reused for another external Transaction.

        } else { // Do not chain.
            // Make sure the external Transaction cannot reach this internal one.
            transaction.internalTransaction = objectManagerState.dummyInternalTransaction;
            // This may not be sufficient to prevent its being used if the external Transaction already has passed
            // the point where it has picked up the referenece to the Internaltransaction. Hence we terminate the
            // InternalTransaction if it is preemptively backed out. See ObjectManagerstate.performCheckpoint()
            // where transactions are backed out without holding the internal transaction synchronize lock.
            if (transactionReference != null)
                transactionReference.clear(); // Inhibt an enqueue of the transactionReference.
            // If a reference is enqueued we are already in inactive state for this
            // transactionReferenceSequence so no action will occur.

            // Tell the ObjectManager that we no longer exist as an active transaction.
            objectManagerState.deRegisterTransaction(this);

        } // if (reUse).

        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.exit(this, cclass, methodName);
    }
447973
@SuppressWarnings("unchecked")
  public CompletableFuture<A> connect() {
    return registry.createPrimitive(name(), type())
        .thenApply(v -> (A) this);
  }
269155
CompletableFuture<Boolean> actualRemoveMaxIdleExpireEntry(K key, V value, long maxIdle, boolean skipLocking) {
      CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
      Object expiringObject = expiring.putIfAbsent(key, completableFuture);
      if (expiringObject == null) {
         if (trace) {
            log.tracef("Submitting expiration removal for key %s which had maxIdle of %s", toStr(key), maxIdle);
         }
         completableFuture.whenComplete((b, t) -> expiring.remove(key, completableFuture));
         try {
            AdvancedCache<K, V> cacheToUse = skipLocking ? cache.withFlags(Flag.SKIP_LOCKING) : cache;
            CompletableFuture<Boolean> expired = cacheToUse.removeMaxIdleExpired(key, value);
            expired.whenComplete((b, t) -> {
               if (t != null) {
                  completableFuture.completeExceptionally(t);
               } else {
                  completableFuture.complete(b);
               }
            });
            return completableFuture;
         } catch (Throwable t) {
            completableFuture.completeExceptionally(t);
            throw t;
         }
      } else if (expiringObject instanceof CompletableFuture) {
         // This means there was another thread that found it had expired via max idle
         return (CompletableFuture<Boolean>) expiringObject;
      } else {
         // If it wasn't a CompletableFuture we had a lifespan removal occurring so it will be removed for sure
         return CompletableFutures.completedTrue();
      }
   }
906007
private void complete(final InstallationManager.InstallationModification modification, final FinalizeCallback callback) {
        final List<File> processed = new ArrayList<File>();
        List<File> reenabled = Collections.emptyList();
        List<File> disabled = Collections.emptyList();
        try {
            try {
                // Update the state to invalidate and process module resources
                if (stateUpdater.compareAndSet(this, State.PREPARED, State.INVALIDATE)) {
                    if (mode == PatchingTaskContext.Mode.APPLY) {
                        // Only invalidate modules when applying patches; on rollback files are immediately restored
                        for (final File invalidation : moduleInvalidations) {
                            processed.add(invalidation);
                            PatchModuleInvalidationUtils.processFile(this, invalidation, mode);
                        }
                        if (!modulesToReenable.isEmpty()) {
                            reenabled = new ArrayList<File>(modulesToReenable.size());
                            for (final File path : modulesToReenable) {
                                reenabled.add(path);
                                PatchModuleInvalidationUtils.processFile(this, path, PatchingTaskContext.Mode.ROLLBACK);
                            }
                        }
                    } else if(mode == PatchingTaskContext.Mode.ROLLBACK) {
                        if (!modulesToDisable.isEmpty()) {
                            disabled = new ArrayList<File>(modulesToDisable.size());
                            for (final File path : modulesToDisable) {
                                disabled.add(path);
                                PatchModuleInvalidationUtils.processFile(this, path, PatchingTaskContext.Mode.APPLY);
                            }
                        }
                    }

                }
                modification.complete();
                callback.completed(this);
                state = State.COMPLETED;
            } catch (Exception e) {
                this.moduleInvalidations.clear();
                this.moduleInvalidations.addAll(processed);
                this.modulesToReenable.clear();
                this.modulesToReenable.addAll(reenabled);
                this.modulesToDisable.clear();
                this.moduleInvalidations.addAll(disabled);
                throw new RuntimeException(e);
            }
        } finally {
            if (state != State.COMPLETED) {
                try {
                    modification.cancel();
                } finally {
                    try {
                        undoChanges();
                    } finally {
                        callback.operationCancelled(this);
                    }
                }
            } else  {
                try {
                    if (checkForGarbageOnRestart) {
                        final File cleanupMarker = new File(installedImage.getInstallationMetadata(), "cleanup-patching-dirs");
                        cleanupMarker.createNewFile();
                    }
                    storeFailedRenaming();
                } catch (IOException e) {
                    PatchLogger.ROOT_LOGGER.debugf(e, "failed to create cleanup marker");
                }
            }
        }
    }
410297
private static <T> BiConsumer<T, Throwable> completeHandler(CompletableCompletionStage<T> s) {
        return (result, failure) -> {
            if (failure == null) {
                s.complete(result);
            } else {
                handleFailure(s, failure);
            }
        };
    }
896604
@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;
	}