Collect {@code <T>} into a buffer, and invoke the given callback whenever
the buffer is full, during an explicit commit, or on close.
320449
public static <T> BatchingCallback<T> batchInto(int size, Callback<? super List<T>> out)
    {
        return new BatchingCallback<T>(size, out);
    }
781221
private synchronized boolean waitForContent()
        throws InterruptedIOException
    {
        if (_buffer!=null)
        {
            if (_buffer.hasRemaining())
                return true;
             
             // recycle buffer
             recycle(_buffer);
             _buffer=null;
        }        
        
        while(!_closed && LazyList.size(_buffers)==0)
        {
            try
            {
                this.wait(_timeout);
            }
            catch(InterruptedException e)
            {
                log.debug(e);
                throw new InterruptedIOException(e.toString());
            }
        }    
        
        if (_closed)
            return false;
            
        if (LazyList.size(_buffers)==0)
            throw new SocketTimeoutException();
        
        _buffer=(ByteBuffer)LazyList.get(_buffers, 0);
        _buffers=LazyList.remove(_buffers, 0);
        
        return true;
    }
717161
private void ensureFill() throws JedisConnectionException {
    if (count >= limit) {
      try {
        limit = in.read(buf);
        count = 0;
        if (limit == -1) {
          throw new JedisConnectionException("Unexpected end of stream.");
        }
      } catch (IOException e) {
        throw new JedisConnectionException(e);
      }
    }
  }
187471
@Override
    public void executeEventCallback(Runnable callback) {
        if (!nodeEngine.isRunning()) {
            return;
        }
        try {
            eventExecutor.execute(callback);
        } catch (RejectedExecutionException e) {
            rejectedCount.inc();

            if (eventExecutor.isLive()) {
                logFailure("EventQueue overloaded! Failed to execute event callback: %s", callback);
            }
        }
    }
1167636
@Override
    public void submitToMember(Runnable command, Member member, ExecutionCallback callback) {
        Callable<?> callable = createRunnableAdapter(command);
        submitToMember(callable, member, callback);
    }
1361995
public <T> T invoke(Callable<T> c) throws Exception {
        try {
            T result = c.call();
            succeed();
            return result;
        } catch (Exception e) {
            fail();
            throw e;
        }
    }
20126
public <T> T act(ProcessCallable<T> callable) throws IOException, InterruptedException {
            return callable.invoke(this, FilePath.localChannel);
        }
244082
public static <T> void exceptionListener(CompletableFuture<T> completableFuture, Consumer<Throwable> exceptionListener) {
        completableFuture.whenComplete((r, ex) -> {
            if (ex != null) {
                Callbacks.invokeSafely(exceptionListener, ex, null);
            }
        });
    }
72576
<V> V execute(final TransactionalClosureWithReturn<V> closure) {
        return getSessionFactory().doInSession(
                new Callable<V>() {
                    @Override
                    public V call() {
                        PersistenceSession persistenceSession = getPersistenceSession();
                        persistenceSession.getServicesInjector().injectServicesInto(closure);
                        return doExecute(closure);
                    }
                }
        );
    }
785698
@Override
  public void accept(final FastFuture<T> t) {

    active.add(t);

    if (active.size() > maxActive.getMaxActive()) {

      while (active.size() > maxActive.getReduceTo()) {

        final List<FastFuture> toRemove = active.stream()
          .filter(cf -> cf.isDone())
          .peek(this::handleExceptions)
          .collect(Collectors.toList());

        active.removeAll(toRemove);
        if (active.size() > maxActive.getReduceTo()) {
          final CompletableFuture promise = new CompletableFuture();
          FastFuture.xOf(active.size() - maxActive.getReduceTo(), () -> promise.complete(true), active.toArray(new FastFuture[0]));

          promise.join();
        }

      }
    }

  }
458915
public synchronized void sendRetry(long retry, EventCallback callback) {

        if (open == 0 || shutdown) {
            if (callback != null) {
                callback.failed(this, null, null, null, new ClosedChannelException());
            }
            return;
        }
        queue.add(new SSEData(retry, callback));
        sink.getIoThread().execute(new Runnable() {
            @Override
            public void run() {
                synchronized (ServerSentEventConnection.this) {
                    if (pooled == null) {
                        fillBuffer();
                        writeListener.handleEvent(sink);
                    }
                }
            }
        });
    }
213369
public static<T> T      callWithRetry(CuratorZookeeperClient client, Callable<T> proc) throws Exception
    {
        T               result = null;
        RetryLoop       retryLoop = client.newRetryLoop();
        while ( retryLoop.shouldContinue() )
        {
            try
            {
                client.internalBlockUntilConnectedOrTimedOut();

                result = proc.call();
                retryLoop.markComplete();
            }
            catch ( Exception e )
            {
                ThreadUtils.checkInterrupted(e);
                retryLoop.takeException(e);
            }
        }
        return result;
    }
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);
	}
360469
@VisibleForTesting
    protected void executionComplete(Throwable exception) {
        Collection<QueueItem> toFail = null;
        Throwable failEx = null;
        synchronized (this.stateLock) {
            // Release the spot occupied by this item's execution.
            this.activeCount--;
            if (exception != null && !this.closed) {
                // Need to fail all future items and close to prevent new items from being processed.
                failEx = new ProcessingException("A previous item failed to commit. Cannot process new items.", exception);
                toFail = new ArrayList<>(this.pendingItems);
                this.pendingItems.clear();
                this.closed = true;
            }

            if (this.emptyNotifier != null && this.activeCount == 0 && this.pendingItems.isEmpty()) {
                // We were asked to notify when we were empty.
                this.emptyNotifier.release();
                this.emptyNotifier = null;
            }
        }

        if (toFail != null) {
            for (QueueItem q : toFail) {
                q.result.completeExceptionally(failEx);
            }

            return;
        }

        // We need to ensure the items are still executed in order. Once out of the main sync block, it is possible that
        // this callback may be invoked concurrently after various completions. As such, we need a special handler for
        // these, using its own synchronization, different from the main one, so that we don't hold that stateLock for too long.
        synchronized (this.processingLock) {
            while (true) {
                QueueItem toProcess;
                synchronized (this.stateLock) {
                    if (hasCapacity() && !this.pendingItems.isEmpty()) {
                        // We have spare capacity and we have something to process. Dequeue it and reserve the spot.
                        toProcess = this.pendingItems.pollFirst();
                        this.activeCount++;
                    } else {
                        // No capacity left or no more pending items.
                        break;
                    }
                }

                Futures.completeAfter(() -> processInternal(toProcess.data), toProcess.result);
            }
        }
    }
943613
private void postProcess(ProcessMethodCallback callback, TransactionContext txContext,
                           InputDatum input, ProcessMethod.ProcessResult result) {
    InputContext inputContext = input.getInputContext();
    Throwable failureCause = null;
    FailureReason.Type failureType = FailureReason.Type.IO_ERROR;
    try {
      if (result.isSuccess()) {
        // If it is a retry input, force the dequeued entries into current transaction.
        if (input.getRetry() > 0) {
          input.reclaim();
        }
        txContext.finish();
      } else {
        failureCause = result.getCause();
        failureType = FailureReason.Type.USER;
        txContext.abort();
      }
    } catch (Throwable e) {
      LOG.error("Transaction operation failed: {}", e.getMessage(), e);
      failureType = FailureReason.Type.IO_ERROR;
      if (failureCause == null) {
        failureCause = e;
      }
      try {
        if (result.isSuccess()) {
          txContext.abort();
        }
      } catch (Throwable ex) {
        LOG.error("Fail to abort transaction: {}", inputContext, ex);
      }
    }

    try {
      if (failureCause == null) {
        callback.onSuccess(result.getEvent(), inputContext);
      } else {
        callback.onFailure(result.getEvent(), inputContext,
                           new FailureReason(failureType, failureCause.getMessage(), failureCause),
                           createInputAcknowledger(input));
      }
    } catch (Throwable t) {
      LOG.error("Failed to invoke callback.", t);
    }
  }
939367
@Override
  protected void flushAndSync(boolean durable) throws IOException {
    if (outputStream == null) {
      throw new IOException("Trying to use aborted output stream!");
    }
    if (doubleBuf.isFlushed()) {
      return;
    }
    // Under the covers, this invokes a method that does not return until
    // we have successfully persisted the buffer to a quorum of bookies.
    doubleBuf.flushTo(outputStream);
  }
76317
public static <T> void stream(Callback<T> callback, Iterable<T> iterable) throws Exception
    {
        for (T item : iterable) {
            try {
                callback.call(item);
            } catch (CallbackRefusedException e) {
                return;
            }
        }
    }