{"record":{"id":"3ea5cbeeef9cd066","repo":"apache/druid","slug":"invalid-operation-kinesisrecordsupplier-has-alre","errorCode":null,"errorMessage":"Invalid operation - KinesisRecordSupplier has already been closed","messagePattern":"Invalid operation - KinesisRecordSupplier has already been closed","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java","lineNumber":1068,"sourceCode":"    }\n\n    String shardIterator = kinesis.getShardIterator(iteratorRequestBuilder.build()).shardIterator();\n\n    return kinesis.getRecords(\n        GetRecordsRequest.builder()\n            .shardIterator(shardIterator)\n            .limit(1)\n            .build()\n    );\n  }\n\n  /**\n   * Explode if {@link #close()} has been called on the supplier.\n   */\n  private void checkIfClosed()\n  {\n    if (closed) {\n      throw new ISE(\"Invalid operation - KinesisRecordSupplier has already been closed\");\n    }\n  }\n\n  /**\n   * This method must be called before a seek operation ({@link #seek}, {@link #seekToLatest}, or\n   * {@link #seekToEarliest}).\n   * <p>\n   * When called, it will nuke the {@link #scheduledExec} that is shared by all {@link PartitionResource}, filters\n   * records from the buffer for partitions which will have a seek operation performed, and stops background fetch for\n   * each {@link PartitionResource} to prepare for the seek. If background fetch is not currently running, the\n   * {@link #scheduledExec} will not be re-created.\n   */\n  private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> partitions) throws InterruptedException\n  {\n    checkIfClosed();\n    if (backgroundFetchEnabled && partitionsFetchStarted.compareAndSet(true, false)) {\n      scheduledExec.shutdown();\n","sourceCodeStart":1050,"sourceCodeEnd":1086,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java#L1050-L1086","documentation":"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.","triggerScenarios":"Calling poll, seek, seekToEarliest, seekToLatest, or other checkIfClosed-guarded methods after close() has been called on the KinesisRecordSupplier instance.","commonSituations":"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.","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"],"exampleFix":"// before\nsupplier.close();\nsupplier.poll(1000);\n// after\nsupplier.close();\n// create a fresh supplier for any further reads\nKinesisRecordSupplier newSupplier = new KinesisRecordSupplier(kinesis, ...);","handlingStrategy":"type-guard","validationCode":"if (supplier.isClosed()) { return; } // skip work after close","typeGuard":"static boolean usable(KinesisRecordSupplier s) { return !s.isClosed(); }","tryCatchPattern":"try {\n  records = supplier.poll(timeoutMs);\n} catch (ISE e) {\n  if (e.getMessage().contains(\"has already been closed\")) {\n    return Collections.emptyList(); // graceful shutdown path\n  }\n  throw e;\n}","preventionTips":["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"],"tags":["kinesis","closed-resource","lifecycle"],"backgroundTag":"invalid-state-transition","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}