code: void cleanupQueue(SubscriptionImpl subscription, Queue<Object> queue) {
          final Throwable cause = ClosedPublisherException.get();
          for (;;) {
              final Object e = queue.poll();
              if (e == null) {
                  break;
              }
  
              try {
                  if (e instanceof CloseEvent) {
                      notifySubscriberOfCloseEvent(subscription, (CloseEvent) e);
                      continue;
                  }
  
                  if (e instanceof CompletableFuture) {
                      ((CompletableFuture<?>) e).completeExceptionally(cause);
                  }
  
                  @SuppressWarnings("unchecked")
                  final T obj = (T) e;
                  onRemoval(obj);
              } finally {
                  ReferenceCountUtil.safeRelease(e);
              }
          }
      }
docs: Helper method for the common case of cleaning up all elements in a queue when shutting down the stream.