@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);
            }
        }
    }
@ VisibleForTesting protected void executionComplete ( Throwable exception ) { Collection < QueueItem > toFail = null ; Throwable failEx = null ; synchronized ( this . stateLock ) { this . activeCount -- ; if ( exception != null && ! this . closed ) { 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 ( ) ) { this . emptyNotifier . release ( ) ; this . emptyNotifier = null ; } } if ( toFail != null ) { for ( QueueItem q : toFail ) { q . result . completeExceptionally ( failEx ) ; } return ; } synchronized ( this . processingLock ) { while ( true ) { QueueItem toProcess ; synchronized ( this . stateLock ) { if ( hasCapacity ( ) && ! this . pendingItems . isEmpty ( ) ) { toProcess = this . pendingItems . pollFirst ( ) ; this . activeCount ++ ; } else { break ; } } Futures . completeAfter ( ( ) -> processInternal ( toProcess . data ) , toProcess . result ) ; } } }
Callback that is invoked when an item has completed execution.

@param exception (Optional) An exception from the execution. If set, it indicates the item has not been
processed successfully.