{"record":{"id":"bacf805aa000e0e0","repo":"apache/druid","slug":"got-null-sequence-number-for-partition-s-when-fe","errorCode":null,"errorMessage":"got null sequence number for partition[%s] when fetching from kafka!","messagePattern":"got null sequence number for partition\\[(.+?)\\] when fetching from kafka!","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java","lineNumber":136,"sourceCode":"      TaskToolbox taskToolbox\n  ) throws InterruptedException, IOException\n  {\n    final String stream = task.getIOConfig().getStartSequenceNumbers().getStream();\n    final boolean isMultiTopic = task.getIOConfig().isMultiTopic();\n    final Map<TopicPartition, Long> resetPartitions = new HashMap<>();\n    boolean doReset = false;\n    if (task.getTuningConfig().isResetOffsetAutomatically()) {\n      for (Map.Entry<TopicPartition, Long> outOfRangePartition : outOfRangePartitions.entrySet()) {\n        final TopicPartition topicPartition = outOfRangePartition.getKey();\n        final long nextOffset = outOfRangePartition.getValue();\n        // seek to the beginning to get the least available offset\n        StreamPartition<KafkaTopicPartition> streamPartition = StreamPartition.of(\n            stream,\n            new KafkaTopicPartition(isMultiTopic, topicPartition.topic(), topicPartition.partition())\n        );\n        final Long leastAvailableOffset = recordSupplier.getEarliestSequenceNumber(streamPartition);\n        if (leastAvailableOffset == null) {\n          throw new ISE(\n              \"got null sequence number for partition[%s] when fetching from kafka!\",\n              topicPartition.partition()\n          );\n        }\n        // reset the seek\n        recordSupplier.seek(streamPartition, nextOffset);\n        // Reset consumer offset if resetOffsetAutomatically is set to true\n        // and the current message offset in the kafka partition is more than the\n        // next message offset that we are trying to fetch\n        if (leastAvailableOffset > nextOffset) {\n          doReset = true;\n          resetPartitions.put(topicPartition, nextOffset);\n        }\n      }\n    }\n\n    if (doReset) {\n      sendResetRequestAndWait(CollectionUtils.mapKeys(resetPartitions, topicPartition -> StreamPartition.of(","sourceCodeStart":118,"sourceCodeEnd":154,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java#L118-L154","documentation":"KafkaIndexTaskRunner.possiblyResetOffsetsOrWait() calls recordSupplier.getEarliestSequenceNumber() to verify a partition still exists; a null return means Kafka reported no offsets for the partition, so the runner throws ISE rather than proceeding with invalid offsets.","triggerScenarios":"During getRecords, when the task (re)reads offsets and getEarliestSequenceNumber returns null — typically the topic/partition was deleted, or the broker has no data for the partition.","commonSituations":"Topic deleted or recreated while a task is running, partition removal, retention deleting all records, misconfigured multi-topic patterns.","solutions":["Verify the topic and its partitions still exist (kafka-topics --describe)","Recreate the topic or restore the deleted partition","Reset the task/datasource offsets to valid values (reset supervisor offsets)","Check broker retention and min.insync settings"],"exampleFix":"// before\ntopic deleted while supervisor running\n// after\nkafka-topics.sh --create --topic mytopic ... ; then kafka-indexing-service reset supervisor offsets","handlingStrategy":"retry","validationCode":"// before/at task config time, confirm partitions exist\ntry (AdminClient admin = AdminClient.create(consumerProps)) {\n  Set<String> topics = admin.listTopics().names().get(30, TimeUnit.SECONDS);\n  if (!topics.containsAll(configuredTopics)) throw new IllegalStateException(\"topic missing\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  records = taskRunner.getRecords(partition, ...);\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"got null sequence number\")) {\n    // topic/partition deleted: recreate topic or reset offsets, then resume\n  } else { throw e; }\n}","preventionTips":["Disable topic auto-deletion in environments running Druid tasks","Monitor for topic deletion (Kafka alerts)","Use offset reset policies and supervisor reset APIs","Verify retention doesn't fully expire partitions"],"tags":["kafka","offsets","partitions"],"backgroundTag":"record-not-found","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"}