/*
(non-Javadoc)

@see
com.netflix.evcache.operation.EVCacheLatchI#onComplete(net.spy.memcached.internal.OperationFuture)
426743
@Override
    public void onComplete(OperationFuture<?> future) throws Exception {
        if (log.isDebugEnabled()) log.debug("BEGIN : onComplete - Calling Countdown. Completed Future = " + future + "; App : " + appName); 
        countDown();
        completeCount++;
        if(getCompletedCount() >= getExpectedSuccessCount()) {
            final String cachePrefix = null;
            EVCacheMetricsFactory.getStatsTimer(getAppName(), cachePrefix, "LatchPolicyDuration").record(System.currentTimeMillis() - startTimeMS);
            if (log.isDebugEnabled()) log.debug("Future policy satisfied. Last Completed Future = " + future + ". Took " + (System.currentTimeMillis() - startTimeMS) + " milliseconds"); 
        }
        EVCacheMetricsFactory.increment(appName, null, "EVCacheLatchImpl-OnComplete");
        if(evcacheEvent != null) {
            if (log.isDebugEnabled()) log.debug(";App : " + evcacheEvent.getAppName() + "; Call : " + evcacheEvent.getCall() + "; Keys : " + evcacheEvent.getEVCacheKeys() + "; completeCount : " + completeCount + "; totalFutureCount : " + totalFutureCount +"; failureCount : " + failureCount);
            try {
                if(future.isDone() && future.get().equals(Boolean.FALSE)) {
                    failureCount++;
                }
            } catch (Exception e) {
                failureCount++;
                if(log.isDebugEnabled()) log.debug(e.getMessage(), e);
            }
            if(!onCompleteDone && getCompletedCount() >= getExpectedSuccessCount()) {
                if(evcacheEvent.getClients().size() > 0) {
                    for(EVCacheClient client : evcacheEvent.getClients()) {
                        final List<EVCacheEventListener> evcacheEventListenerList = client.getPool().getEVCacheClientPoolManager().getEVCacheEventListeners();
                        for (EVCacheEventListener evcacheEventListener : evcacheEventListenerList) {
                            evcacheEventListener.onComplete(evcacheEvent);
                        }
                        EVCacheMetricsFactory.increment(evcacheEvent.getAppName(), evcacheEvent.getCacheName(), "EVCacheLatchImpl-OnComplete-Done");
                        onCompleteDone = true;//This ensures we fire onComplete only once
                        break;
                    }
                }
            }
            if(scheduledFuture != null) {
                final boolean futureCancelled = scheduledFuture.isCancelled(); 
                if (log.isDebugEnabled()) log.debug("App : " + evcacheEvent.getAppName() + "; Call : " + evcacheEvent.getCall() + "; Keys : " + evcacheEvent.getEVCacheKeys() + "; completeCount : " + completeCount + "; totalFutureCount : " + totalFutureCount +"; failureCount : " + failureCount + "; futureCancelled : " + futureCancelled);
                if(onCompleteDone && !futureCancelled) {
                    if(completeCount == totalFutureCount && failureCount == 0) { // all futures are completed
                        final boolean status = scheduledFuture.cancel(true);
                        if (log.isDebugEnabled()) log.debug("Cancelled the scheduled task : " + status);
                        if(status) {
                            EVCacheMetricsFactory.increment(evcacheEvent.getAppName(), evcacheEvent.getCacheName(), "EVCacheLatchImpl-OnComplete-FutureUnregistered-SUCCESS");
                        } else {
                            EVCacheMetricsFactory.increment(evcacheEvent.getAppName(), evcacheEvent.getCacheName(), "EVCacheLatchImpl-OnComplete-FutureUnregistered-FAIL");
                        }
                    }
                }
            }
            if (log.isDebugEnabled()) log.debug("App : " + evcacheEvent.getAppName() + "; Call : " + evcacheEvent.getCall() + "; Keys : " + evcacheEvent.getEVCacheKeys() + "; completeCount : " + completeCount + "; totalFutureCount : " + totalFutureCount +"; failureCount : " + failureCount);
        }
        if (log.isDebugEnabled()) log.debug("END : onComplete - Calling Countdown. Completed Future = " + future + "; App : " + appName); 
    }
637697
@Override
    public void complete(VirtualConnection vc, TCPWriteRequestContext wsc) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "complete, vc=" + getVCHash());
        }

        synchronized (closeSync) {
            if (closeCalled) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
                    Tr.exit(tc, "complete");
                }
                return;
            }
        }

        VirtualConnection rc = vc;
        if (0 < this.asyncBytesToWrite) {
            // previously went async partly through the app data, continue now
            rc = encryptAndWriteAsync(this.asyncBytesToWrite, false, this.asyncTimeout);
        }

        if (null != rc) {
            // Nothing else needs to be done here. Report success up the chain.
            this.callback.complete(rc, this);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "complete");
        }
    }
1199908
@Override
    public CompletableFuture<Void> add(Operation operation, Duration timeout) {
        ensureRunning();
        return this.operationProcessor.process(operation);
    }
108490
@Override
    public void awaitCompletion() {
        while (!hydratedQueue.isEmpty()) {
            HydratedPojo<?> hydratedPojo = hydratedQueue.remove();
            if(!Futures.getUnchecked(hydratedPojo.getFuture())) {
                throw new PojoNotFoundException(hydratedPojo.getBinding(), hydratedPojo.getTableName(), hydratedPojo.getKeys());
            }
        }
    }
931587
void onCompletion(Collection<ApplicationDependency> dependencies) {
        if (!dependencies.isEmpty()) {
            for (ApplicationDependency dependency : dependencies) {
                dependency.onCompletion(this);
            }
        }
    }
350259
public synchronized void setComplete(boolean completed) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setComplete", completed);
        this.completed = completed;
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setComplete");
    }
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());
    }
  }
1194835
@Override
    public void operationComplete(ChannelFuture future) throws Exception {
        // If a message has been sent out, cancel the timeout for starting a request.
        cancelTimeout();

        if (future.isSuccess()) {
            // The first write is always the first headers, so log that we finished our first transfer over the
            // wire.
            if (!loggedRequestFirstBytesTransferred) {
                logBuilder.requestFirstBytesTransferred();
                loggedRequestFirstBytesTransferred = true;
            }

            if (state == State.DONE) {
                // Successfully sent the request; schedule the response timeout.
                response.scheduleTimeout(ch.eventLoop());
            }

            // Request more messages regardless whether the state is DONE. It makes the producer have
            // a chance to produce the last call such as 'onComplete' and 'onError' when there are
            // no more messages it can produce.
            if (!isSubscriptionCompleted) {
                assert subscription != null;
                subscription.request(1);
            }
            return;
        }

        fail(future.cause());

        final Throwable cause = future.cause();
        if (!(cause instanceof ClosedPublisherException)) {
            final Channel ch = future.channel();
            Exceptions.logIfUnexpected(logger, ch, HttpSession.get(ch).protocol(), cause);
            ch.close();
        }
    }
444014
@Override
  public void onCompleted() {
    RpcUtils.streamingRPCAndLog(LOG, new RpcUtils.StreamingRpcCallable<OpenLocalBlockResponse>() {
      @Override
      public OpenLocalBlockResponse call() throws Exception {
        if (mLockId != BlockLockManager.INVALID_LOCK_ID) {
          mWorker.unlockBlock(mLockId);
          mLockId = BlockLockManager.INVALID_LOCK_ID;
        } else if (mRequest != null) {
          LOG.warn("Close a closed block {}.", mRequest.getBlockId());
        }
        return null;
      }

      @Override
      public void exceptionCaught(Throwable e) {
        mResponseObserver.onError(GrpcExceptionUtils.fromThrowable(e));
        mLockId = BlockLockManager.INVALID_LOCK_ID;
      }
    }, "CloseBlock", false, true, mResponseObserver, "Session=%d, Request=%s",
        mSessionId, mRequest);
  }
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;
    }
836973
public static <T> void completeFrom(CompletionStage<T> source, CompletableFuture<T> target) {
    source.whenComplete(
        (t, error) -> {
          if (error != null) {
            target.completeExceptionally(error);
          } else {
            target.complete(t);
          }
        });
  }
360597
@Override
   public CompletableFuture<Object> invokeAsync() throws Throwable {
      // make sure the get command doesn't perform a remote call
      // as our caller is already calling the ClusteredGetCommand on all the relevant nodes
      // CACHE_MODE_LOCAL is not used as it can be used when we want to ignore the ownership with respect to reads
      long flagBitSet = EnumUtil.bitSetOf(Flag.SKIP_REMOTE_LOOKUP);
      GetCacheEntryCommand command = commandsFactory.buildGetCacheEntryCommand(key, segment,
            EnumUtil.mergeBitSets(flagBitSet, getFlagsBitSet()));
      command.setTopologyId(topologyId);
      InvocationContext invocationContext = icf.createRemoteInvocationContextForCommand(command, getOrigin());
      CompletableFuture<Object> future = invoker.invokeAsync(invocationContext, command);
      return future.thenApply(rv -> {
         if (trace) log.tracef("Return value for key=%s is %s", key, rv);
         //this might happen if the value was fetched from a cache loader
         if (rv instanceof MVCCEntry) {
            MVCCEntry mvccEntry = (MVCCEntry) rv;
            return entryFactory.createValue(mvccEntry);
         } else if (rv instanceof InternalCacheEntry) {
            InternalCacheEntry internalCacheEntry = (InternalCacheEntry) rv;
            return internalCacheEntry.toInternalCacheValue();
         } else { // null or Response
            return rv;
         }
      });
   }
1363541
synchronized void onComplete(@javax.annotation.Nullable Throwable error) {
    if (isCompleted()) {
      return;
    }
    currentConfigObserver = null;
    // TODO(songya): add Runnable
    Status status;
    if (error == null) {
      status = Status.OK;
    } else if (error instanceof StatusRuntimeException) {
      status = ((StatusRuntimeException) error).getStatus();
    } else {
      status = Status.UNKNOWN;
    }
    terminateStatus = status;
  }
859360
private void onNextAttemptStarted(CompletionStage<Boolean> futureOutcome) {
    assert executor.inEventLoop();
    state = State.ATTEMPT_IN_PROGRESS;
    futureOutcome
        .whenCompleteAsync(this::onNextAttemptCompleted, executor)
        .exceptionally(UncaughtExceptions::log);
  }
1183533
private void updateNetwork(final NetworkContext context, final Handler<AsyncResult<Void>> doneHandler) {
    final CountingCompletionHandler<Void> complete = new CountingCompletionHandler<Void>(context.components().size());
    complete.setHandler(new Handler<AsyncResult<Void>>() {
      @Override
      public void handle(AsyncResult<Void> result) {
        if (result.failed()) {
          new DefaultFutureResult<Void>(result.cause()).setHandler(doneHandler);
        } else {
          new DefaultFutureResult<Void>((Void) null).setHandler(doneHandler);
        }
      }
    });
    updateComponents(context.components(), complete);
  }
1022763
private void onAutoCompleted(final Event event, final Suggestion<T> suggestion) {
        TypeaheadAutoCompletedEvent.fire(this, suggestion, event);
    }
1018225
@Override
    public CompletableFuture<Void> setAsync(int expireSeconds, String key, V value) {
        return (CompletableFuture) setAsync(key, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
    }