apache/druid · error · IllegalStateException

Closed

Error message

Closed

What it means

AccumulatingProcessorManager wraps a delegate ProcessorManager with a lookahead (currentResult). Once the manager has been fully consumed, currentResult is null and next() can no longer be called; invoking it then is an invalid state, so it throws ISE('Closed'). It indicates the iterator-style manager was advanced past its end and used again.

Solutions

1. Stop calling next() once it returns Optional.empty()
2. Create a fresh AccumulatingProcessorManager for each processing pass
3. Restructure the consumer loop so next() is only invoked while the previous result was present

Example fix

// before
while (true) {
  Optional<ProcessorAndCallback<T>> cb = manager.next().get();
  handle(cb);
}
// after
while (manager.next().get().isPresent()) {
  handle(manager.next().get()); // or store result before handling
}

if (!manager.hasNextState()) { return; } // track consumption yourself before calling next()

null

try { Optional<ProcessorAndCallback<T>> r = manager.next().get(); if (!r.isPresent()) { done = true; } } catch (IllegalStateException e) { /* manager consumed: recreate manager */ }

When it happens

Trigger: Calling next() after the manager has already returned the final Optional.empty / completed its iteration (currentResult set to null), typically by looping beyond the completion condition.

Common situations: Custom processor loops that don't check the returned Optional before calling next() again; reusing a consumed manager instance for a second pass instead of creating a new one.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Cannot add merged batches for level
• Cannot start once already started or closed
• FrameTooLarge
• No await set
• No channels set

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/a5314e09dc302551. Report an issue: GitHub.

Appendix: source

Thrown at processing/src/main/java/org/apache/druid/frame/processor/manager/AccumulatingProcessorManager.java:54

  private final BiFunction<R, T, R> accumulateFn;
  private R currentResult;

  public AccumulatingProcessorManager(
      ProcessorManager<T, ?> delegate,
      R initialResult,
      BiFunction<R, T, R> accumulateFn
  )
  {
    this.delegate = delegate;
    this.currentResult = Preconditions.checkNotNull(initialResult, "initialResult");
    this.accumulateFn = accumulateFn;
  }

  @Override
  public ListenableFuture<Optional<ProcessorAndCallback<T>>> next()
  {
    if (currentResult == null) {
      throw new ISE("Closed");
    }

    return FutureUtils.transform(
        delegate.next(),
        nextProcessor -> nextProcessor.map(
            retVal -> new ProcessorAndCallback<>(
                retVal.processor(),
                r -> {
                  currentResult = accumulateFn.apply(currentResult, r);
                  retVal.onComplete(r);
                }
            )
        )
    );
  }

  @Override
  public R result()

View on GitHub (pinned to 9b90983fd2)