code: @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);
              }
          }
      }
docs: 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.