handle the asynchronous completable future flow

@param proceedingJoinPoint AOPJoinPoint
@param bulkhead            configured bulkhead
@return CompletionStage
776120
private Object handleJoinPointCompletableFuture(ProceedingJoinPoint proceedingJoinPoint, io.github.resilience4j.bulkhead.Bulkhead bulkhead) {
		return bulkhead.executeCompletionStage(() -> {
			try {
				return (CompletionStage<?>) proceedingJoinPoint.proceed();
			} catch (Throwable throwable) {
				throw new CompletionException(throwable);
			}
		});
	}
201904
public void onComplete(
			final Consumer<StreamElementQueueEntry<T>> completeFunction,
			Executor executor) {
		final StreamElementQueueEntry<T> thisReference = this;

		getFuture().whenCompleteAsync(
			// call the complete function for normal completion as well as exceptional completion
			// see FLINK-6435
			(value, throwable) -> completeFunction.accept(thisReference),
			executor);
	}
262228
public static <T> Publisher<T> onComplete(Publisher<T> publisher, Supplier<CompletableFuture<Void>> future) {
        return actual -> publisher.subscribe(new CompletionAwareSubscriber<T>() {
            @Override
            protected void doOnSubscribe(Subscription subscription) {
                actual.onSubscribe(subscription);
            }

            @Override
            protected void doOnNext(T message) {
                try {
                    actual.onNext(message);
                } catch (Throwable e) {
                    onError(e);
                }

            }

            @Override
            protected void doOnError(Throwable t) {
                actual.onError(t);
            }

            @Override
            protected void doOnComplete() {
                future.get().whenComplete((aVoid, throwable) -> {
                    if (throwable != null) {
                        actual.onError(throwable);
                    } else {
                        actual.onComplete();
                    }
                });
            }
        });
    }
1190000
@Override
	public CompletableFuture<Acknowledge> updateTaskExecutionState(
			final TaskExecutionState taskExecutionState) {
		checkNotNull(taskExecutionState, "taskExecutionState");

		if (executionGraph.updateState(taskExecutionState)) {
			return CompletableFuture.completedFuture(Acknowledge.get());
		} else {
			return FutureUtils.completedExceptionally(
				new ExecutionGraphException("The execution attempt " +
					taskExecutionState.getID() + " was not found."));
		}
	}
1363867
private CompletionStage<Void> emitMembershipUpdateEvent() {
        final IRI membershipResource = parent.getMembershipResource().map(MutatingLdpHandler::removeHashFragment)
            .orElse(null);
        if (membershipResource != null) {
            return allOf(getServices().getResourceService().touch(membershipResource).toCompletableFuture(),
                getServices().getResourceService().get(membershipResource).thenAccept(res -> {
                    if (res.getIdentifier() != null) {
                        getServices().getEventService().emit(new SimpleEvent(getUrl(res.getIdentifier()),
                                    getSession().getAgent(), asList(PROV.Activity, AS.Update),
                                    asList(res.getInteractionModel())));
                    }
                }).toCompletableFuture());
        }
        return completedFuture(null);
    }
395438
public static <T> BulkheadOperator<T> of(Bulkhead bulkhead) {
        return of(bulkhead, Schedulers.parallel());
    }
10963
protected <T extends OperationResponse> void completeOperation(OperationResult result, OperationResponse.Builder<?, T> builder, Throwable error, CompletableFuture<T> future) {
    if (result != null) {
      builder.withIndex(result.index());
      builder.withEventIndex(result.eventIndex());
      if (result.failed()) {
        error = result.error();
      }
    }

    if (error == null) {
      if (result == null) {
        future.complete(builder.withStatus(RaftResponse.Status.ERROR)
            .withError(RaftError.Type.PROTOCOL_ERROR)
            .build());
      } else {
        future.complete(builder.withStatus(RaftResponse.Status.OK)
            .withResult(result.result())
            .build());
      }
    } else if (error instanceof CompletionException && error.getCause() instanceof RaftException) {
      future.complete(builder.withStatus(RaftResponse.Status.ERROR)
          .withError(((RaftException) error.getCause()).getType(), error.getMessage())
          .build());
    } else if (error instanceof RaftException) {
      future.complete(builder.withStatus(RaftResponse.Status.ERROR)
          .withError(((RaftException) error).getType(), error.getMessage())
          .build());
    } else if (error instanceof PrimitiveException.ServiceException) {
      log.warn("An application error occurred: {}", error.getCause());
      future.complete(builder.withStatus(RaftResponse.Status.ERROR)
          .withError(RaftError.Type.APPLICATION_ERROR)
          .build());
    } else {
      log.warn("An unexpected error occurred: {}", error);
      future.complete(builder.withStatus(RaftResponse.Status.ERROR)
          .withError(RaftError.Type.PROTOCOL_ERROR, error.getMessage())
          .build());
    }
  }
820984
static <T> CompletionStage<Boolean> authorize(Authorizer<T> authorizer, ServiceRequestContext ctx, T data) {
        try {
            final CompletionStage<Boolean> f = authorizer.authorize(ctx, data);
            if (f == null) {
                throw new NullPointerException("An " + Authorizer.class.getSimpleName() +
                                               " returned null: " + authorizer);
            }
            return f;
        } catch (Throwable cause) {
            return CompletableFutures.exceptionallyCompletedFuture(cause);
        }
    }
1097593
public ISynchronizationPoint<IOException> finishAsync() {
		if (finishing != null) return finishing;
		finishing = new SynchronizationPoint<>();
		Task<Void,NoException> task = new Task.Cpu<Void,NoException>("Finishing zip compression", priority) {
			@Override
			public Void run() {
				AsyncWork<Integer, IOException> lastWrite = null;
				deflater.finish();
				if (!deflater.finished()) {
					do {
						byte[] writeBuf = new byte[8192];
						int nb = deflater.deflate(writeBuf, 0, writeBuf.length);
						if (nb <= 0) break;
						try { lastWrite = writeOps.write(ByteBuffer.wrap(writeBuf, 0, nb)); }
						catch (IOException e) {
							finishing.error(e);
							return null;
						}
					} while (!deflater.finished());
				}
				if (lastWrite == null) lastWrite = writeOps.getLastPendingOperation();
				if (lastWrite != null)
					lastWrite.listenInline(finishing);
				else
					finishing.unblock();
				return null;
			}
		};
		task.start();
		return operation(finishing);
	}
879787
public CompletableFuture<Long> appendEntries(long index) {
    raft.checkThread();

    if (index == 0) {
      return appendEntries();
    }

    if (index <= raft.getCommitIndex()) {
      return CompletableFuture.completedFuture(index);
    }

    // If there are no other stateful servers in the cluster, immediately commit the index.
    if (raft.getCluster().getActiveMemberStates().isEmpty() && raft.getCluster().getPassiveMemberStates().isEmpty()) {
      long previousCommitIndex = raft.getCommitIndex();
      raft.setCommitIndex(index);
      completeCommits(previousCommitIndex, index);
      return CompletableFuture.completedFuture(index);
    }
    // If there are no other active members in the cluster, update the commit index and complete the commit.
    // The updated commit index will be sent to passive/reserve members on heartbeats.
    else if (raft.getCluster().getActiveMemberStates().isEmpty()) {
      long previousCommitIndex = raft.getCommitIndex();
      raft.setCommitIndex(index);
      completeCommits(previousCommitIndex, index);
      return CompletableFuture.completedFuture(index);
    }

    // Only send entry-specific AppendRequests to active members of the cluster.
    return appendFutures.computeIfAbsent(index, i -> {
      for (RaftMemberContext member : raft.getCluster().getActiveMemberStates()) {
        appendEntries(member);
      }
      return new CompletableFuture<>();
    });
  }
1328754
protected boolean completeAppend(boolean succeeded, long lastLogIndex, CompletableFuture<AppendResponse> future) {
    future.complete(logResponse(AppendResponse.builder()
        .withStatus(RaftResponse.Status.OK)
        .withTerm(raft.getTerm())
        .withSucceeded(succeeded)
        .withLastLogIndex(lastLogIndex)
        .build()));
    return succeeded;
  }
953890
@Override
    protected <T> CompletableFuture<T> doExecute(Command<T> command) throws Exception {
        final CompletableFuture<T> future = new CompletableFuture<>();
        executor.execute(() -> {
            try {
                future.complete(blockingExecute(command));
            } catch (Throwable t) {
                future.completeExceptionally(t);
            }
        });
        return future;
    }
148236
public <N> CompletableFutureQueue<T, N> thenCombineAsync(Function<R, N> nextTransformation) {
    final Function<T, CompletableFuture<N>> f = this.resultFunction
        .andThen(before -> before.thenComposeAsync(v -> supplyAsync(() -> nextTransformation.apply(v))));
    return new CompletableFutureQueue<>(f);
  }
534946
@FFDCIgnore(JobStoppingException.class)
    private void executeAndWaitForCompletion(PartitionPlanDescriptor currentPlan) throws JobRestartException {
        if (isStoppingStoppedOrFailed()) {
            logger.fine("Job already in "
                        + runtimeWorkUnitExecution.getWorkUnitJobContext().getBatchStatus().toString()
                        + " state, exiting from executeAndWaitForCompletion() before beginning execution");
            return;
        }

        List<Integer> partitionsToExecute = getPartitionNumbersToExecute(currentPlan);

        logger.fine("Partitions to execute in this run: " + partitionsToExecute
                    + ".  Total number of partitions in step: " + currentPlan.getNumPartitionsInPlan());

        List<Integer> startedPartitions = new ArrayList<Integer>();
        List<Integer> finishedPartitions = new ArrayList<Integer>();
        List<Throwable> analyzerExceptions = new ArrayList<Throwable>();

        // Keep looping until all partitions have finished.
        while (finishedPartitions.size() < partitionsToExecute.size()) {

            startPartitions(partitionsToExecute, startedPartitions, finishedPartitions, currentPlan);

            // Check that there are still un-finished partitions running.
            // If not, break out of the loop.
            if (finishedPartitions.size() >= partitionsToExecute.size()) {
                break;
            }

            //Break this loop when nextPartitionFinished is -1
            try {
                waitForNextPartitionToFinish(analyzerExceptions, finishedPartitions);
            } catch (JobStoppingException e) {
                // break the loop
                break;
            }
        }

        if (!analyzerExceptions.isEmpty()) {
            rollbackPartitionedStep();
            throw new BatchContainerRuntimeException("Exception previously thrown by Analyzer, rolling back step.", analyzerExceptions.get(0));
        }
    }
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);
    }
353878
public void sendBulkDataAsync(List<? extends Map<String, ?>> bulkSourceList,
            String index, String type) {
        executeBulkRequestAsync(
                buildBulkIndexRequestBuilder(bulkSourceList
                        .stream().map(source -> jmESClient
                                .prepareIndex(index, type).setSource(source))
                        .collect(toList())));
    }
1051598
@Restricted(NoExternalUse.class)
    public synchronized final void maybeComplete() {
        assert this.executor != null;
        if (result != null) {
            executor.completedAsynchronous(result != NULL ? result : null);
            result = null;
        }
    }