{"record":{"id":"bd9c5fe226818524","repo":"apache/druid","slug":"partition-s-has-not-been-assigned","errorCode":null,"errorMessage":"Partition [%s] has not been assigned","messagePattern":"Partition \\[(.+?)\\] has not been assigned","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":915,"sourceCode":"  public boolean isAnyFetchActive()\n  {\n    return partitionResources.values()\n                             .stream()\n                             .map(pr -> pr.currentFetch)\n                             .anyMatch(fetch -> (fetch != null && !fetch.isDone()));\n  }\n\n  /**\n   * Check that a {@link PartitionResource} has been assigned to this record supplier, and if so call\n   * {@link PartitionResource#seek} to move it to the latest offsets. Note that this method does not restart background\n   * fetch, which should have been stopped prior to calling this method by a call to\n   * {@link #filterBufferAndResetBackgroundFetch}.\n   */\n  private void partitionSeek(StreamPartition<String> partition, String sequenceNumber, ShardIteratorType iteratorEnum)\n  {\n    PartitionResource resource = partitionResources.get(partition);\n    if (resource == null) {\n      throw new ISE(\"Partition [%s] has not been assigned\", partition);\n    }\n    resource.seek(iteratorEnum, sequenceNumber);\n  }\n\n  /**\n   * Given a partition and a {@link ShardIteratorType}, create a shard iterator and fetch\n   * {@link #GET_SEQUENCE_NUMBER_RECORD_COUNT} records and return the first sequence number from the result set.\n   * This method is thread safe as it does not depend on the internal state of the supplier (it doesn't use the\n   * {@link PartitionResource} which have been assigned to the supplier), and the Kinesis client is thread safe.\n   * <p>\n   * When there are no records at the offset corresponding to the ShardIteratorType,\n   *    If shard is closed, return custom EOS sequence marker\n   *    While getting the earliest sequence number, return a custom marker corresponding to TRIM_HORIZON\n   *    While getting the most recent sequence number, return a custom marker corresponding to LATEST\n   */\n  @Nullable\n  private String getSequenceNumber(StreamPartition<String> partition, ShardIteratorType iteratorEnum)\n  {","sourceCodeStart":897,"sourceCodeEnd":933,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java#L897-L933","documentation":"partitionSeek looks up the PartitionResource for the given StreamPartition and throws this ISE if none has been assigned. A seek (and the public seek/seekToEarliest/seekToLatest paths that route through it) is only legal after the partition was assigned to this supplier.","triggerScenarios":"Calling seek/seekToEarliest/seekToLatest (or partitionSeek directly) with a StreamPartition that was never added via assign/partition assignment, or after assignment was cleared.","commonSituations":"Seeking a shard that was split/merged away and is no longer in the assignment list; task restart where assignment did not include the partition in metadata; typos or stale partitions after stream resharding.","solutions":["Ensure assign(...) (or setStream/partition assignment) is called with the partition before seeking it","List current shards via getPartitionIds and seek only shards present in the assignment","Refresh shard map after resharding — closed shards cannot be seeked by iterator; use their ending sequence number","Verify the partition key (stream name + shard id) matches exactly what was assigned"],"exampleFix":"// before\nsupplier.seek(partition, sequenceNumber);\n// after\nif (assignedPartitions.contains(partition)) {\n  supplier.seek(partition, sequenceNumber);\n} else {\n  LOG.warn(\"Skipping seek for unassigned partition %s\", partition);\n}","handlingStrategy":"validation","validationCode":"Set<StreamPartition<String>> assigned = supplier.getAssignment();\nif (assigned == null || !assigned.contains(partition)) {\n  throw new IllegalArgumentException(\"seek requires partition assigned: \" + partition);\n}","typeGuard":"static boolean canSeek(KinesisRecordSupplier s, StreamPartition<String> p) {\n  return s.getAssignment() != null && s.getAssignment().contains(p);\n}","tryCatchPattern":"try {\n  supplier.seek(partition, seqNum);\n} catch (ISE e) {\n  if (e.getMessage().contains(\"has not been assigned\")) {\n    supplier.assign(partitions); // re-assign then retry\n    supplier.seek(partition, seqNum);\n  } else throw e;\n}","preventionTips":["Always call assign() (directly or via seek setup) before any seek","Re-derive shard lists after stream resharding; skip shards that were merged/closed","Compare StreamPartition equality (stream + shardId) exactly as assigned","Check metadata storage assignment matches the partitions you seek"],"tags":["kinesis","assignment","seek"],"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"}