apache/druid · error · IllegalStateException

Invalid operation - KinesisRecordSupplier has already been c

Error message

Invalid operation - KinesisRecordSupplier has already been closed

What it means

checkIfClosed() is invoked before operations that require a live supplier. Once close() has run (closed=true), any further use throws this ISE. It guards the scheduled executor and shard iterators, which no longer exist after close.

Source

Thrown at extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java:1068

    }

    String shardIterator = kinesis.getShardIterator(iteratorRequestBuilder.build()).shardIterator();

    return kinesis.getRecords(
        GetRecordsRequest.builder()
            .shardIterator(shardIterator)
            .limit(1)
            .build()
    );
  }

  /**
   * Explode if {@link #close()} has been called on the supplier.
   */
  private void checkIfClosed()
  {
    if (closed) {
      throw new ISE("Invalid operation - KinesisRecordSupplier has already been closed");
    }
  }

  /**
   * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or
   * {@link #seekToEarliest}).
   * <p>
   * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters
   * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for
   * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the
   * {@link #scheduledExec} will not be re-created.
   */
  private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException
  {
    checkIfClosed();
    if (backgroundFetchEnabled && partitionsFetchStarted.compareAndSet(true, false)) {
      scheduledExec.shutdown();

View on GitHub (pinned to 9b90983fd2)

Solutions

Do not use the supplier after close(); restructure code so close() is the last operation (try-with-resources)
Guard consumer loops with an isClosed/stop flag checked before each poll
Create a new KinesisRecordSupplier instance if you need to read again after closing
Synchronize close() and read paths if multiple threads share the supplier

Example fix

// before
supplier.close();
supplier.poll(1000);
// after
supplier.close();
// create a fresh supplier for any further reads
KinesisRecordSupplier newSupplier = new KinesisRecordSupplier(kinesis, ...);

Defensive patterns

Strategy: type-guard

Validate before calling

if (supplier.isClosed()) { return; } // skip work after close

Type guard

static boolean usable(KinesisRecordSupplier s) { return !s.isClosed(); }

Try / catch

try {
  records = supplier.poll(timeoutMs);
} catch (ISE e) {
  if (e.getMessage().contains("has already been closed")) {
    return Collections.emptyList(); // graceful shutdown path
  }
  throw e;
}

Prevention

Use try-with-resources or a strict close-last lifecycle for the supplier
Check the closed/stop flag in consumer loops before every poll
Never share one supplier across threads with independent close logic
After close(), create a new instance instead of reusing the old one

When it happens

Trigger: Calling poll, seek, seekToEarliest, seekToLatest, or other checkIfClosed-guarded methods after close() has been called on the KinesisRecordSupplier instance.

Common situations: Using a supplier after task shutdown hooks ran; accidentally closing the supplier in a finally block while a late poll still executes; sharing one supplier across threads where one closes while another reads.

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

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