{"record":{"id":"444ca328265db422","repo":"apache/druid","slug":"exception-while-seeking-to-the-s-offset-of-part","errorCode":null,"errorMessage":"Exception while seeking to the [%s] offset of partitions in topic [%s]: %s","messagePattern":"Exception while seeking to the \\[(.+?)\\] offset of partitions in topic \\[(.+?)\\]: (.+?)","errorType":"exception","errorClass":"SamplerException","httpStatus":null,"severity":"error","filePath":"indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/RecordSupplierInputSource.java","lineNumber":95,"sourceCode":"  private void assignAndSeek(RecordSupplier<PartitionIdType, SequenceOffsetType, RecordType> recordSupplier)\n  {\n    try {\n      final Set<StreamPartition<PartitionIdType>> partitions = recordSupplier\n          .getPartitionIds(topic)\n          .stream()\n          .map(partitionId -> StreamPartition.of(topic, partitionId))\n          .collect(Collectors.toSet());\n\n      recordSupplier.assign(partitions);\n\n      if (useEarliestOffset) {\n        recordSupplier.seekToEarliest(partitions);\n      } else {\n        recordSupplier.seekToLatest(partitions);\n      }\n    }\n    catch (Exception e) {\n      throw new SamplerException(\n          e,\n          \"Exception while seeking to the [%s] offset of partitions in topic [%s]: %s\",\n          useEarliestOffset ? \"earliest\" : \"latest\",\n          topic,\n          Throwables.getRootCause(e).getMessage()\n      );\n    }\n  }\n\n  @Override\n  public boolean isSplittable()\n  {\n    return false;\n  }\n\n  @Override\n  public boolean needsFormat()\n  {","sourceCodeStart":77,"sourceCodeEnd":113,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/RecordSupplierInputSource.java#L77-L113","documentation":"RecordSupplierInputSource.assignAndSeek() positions the record supplier (e.g. Kafka or Kinesis consumer) at the earliest or latest offset of the requested topic/stream partitions for sampling. If any seek operation throws, it wraps the root cause into a SamplerException carrying the offset mode, topic, and root-cause message. This aborts data sampling for supervisor spec ingestion.","triggerScenarios":"Calling the sampling endpoint (POST /druid/indexer/v1/sampler) whose ioConfig has useEarliestSequenceNumber true/false; recordSupplier.seekToEarliest or seekToLatest throws because partitions don't exist, the broker is unreachable, the consumer has no permission, or offsets are out of range.","commonSituations":"Typo in topic name or partition numbers; topic auto-creation disabled and topic absent; Kafka broker auth (SASL/ACLs) denying the consumer; Kinesis shard being split/merged mid-seek; network/firewall issues between the router/overlord and brokers.","solutions":["Read the root-cause message in the exception and fix the underlying broker/connection issue (wrong topic name, missing partitions, auth failure).","Verify the topic/stream exists and the listed partitions are valid: `kafka-topics.sh --describe --topic <topic>`.","Check network connectivity and security config (bootstrap servers, SASL/SSL, Kinesis region/credentials).","Retry sampling after the broker recovers; confirm the supervisor spec's topic and partitions match reality."],"exampleFix":"// before: sampling a non-existent topic\n\"spec\": {\"ioConfig\": {\"topic\": \"kafak-events\", \"partitions\": [0, 1]}}\n// after: corrected topic name\n\"spec\": {\"ioConfig\": {\"topic\": \"kafka-events\", \"partitions\": [0, 1]}}","handlingStrategy":"validation","validationCode":"// verify topic and partitions before sampling\nkafka-topics.sh --bootstrap-server broker:9092 --describe --topic <topic>\n# ensure requested partitions exist and broker is reachable","typeGuard":null,"tryCatchPattern":"try { sample(spec); } catch (SamplerException e) { log.error(\"Sampling failed: {}\", Throwables.getRootCause(e).getMessage()); throw new ValidationException(\"Fix topic/broker config before sampling\", e); }","preventionTips":["Validate topic names and partition lists against the live cluster","Test broker connectivity and SASL/SSL config before sampling","Enable topic existence checks before supervisor creation"],"tags":["kafka","streaming","sampling","offset-seek"],"backgroundTag":"connection-refused","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}