{"record":{"id":"46bb9f033e48410f","repo":"apache/druid","slug":"new-sequence-startoffset-s-does-not-equal-expect","errorCode":null,"errorMessage":"New sequence startOffset[%s] does not equal expected prior offset[%s]","messagePattern":"New sequence startOffset\\[(.+?)\\] does not equal expected prior offset\\[(.+?)\\]","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java","lineNumber":1338,"sourceCode":"  }\n\n  private void addSequence(final SequenceMetadata<PartitionIdType, SequenceOffsetType> sequenceMetadata)\n  {\n    sequencesLock.lock();\n    try {\n      final SequenceMetadata<PartitionIdType, SequenceOffsetType> latestSequence =\n          getLastSequenceMetadataOrNull(sequences);\n\n      // Sanity check that the start of the new sequence matches up with the end of the prior sequence.\n      for (Map.Entry<PartitionIdType, SequenceOffsetType> entry : sequenceMetadata.getStartOffsets().entrySet()) {\n        final PartitionIdType partition = entry.getKey();\n        final SequenceOffsetType startOffset = entry.getValue();\n\n        if (latestSequence != null) {\n          final SequenceOffsetType priorOffset = latestSequence.getEndOffsets().get(partition);\n\n          if (!startOffset.equals(priorOffset)) {\n            throw new ISE(\n                \"New sequence startOffset[%s] does not equal expected prior offset[%s]\",\n                startOffset,\n                priorOffset\n            );\n          }\n        }\n      }\n\n      if (!isEndOffsetExclusive() && latestSequence != null) {\n        if (!latestSequence.getEndOffsets().keySet().equals(sequenceMetadata.getExclusiveStartPartitions())) {\n          throw new ISE(\n              \"Exclusive start partitions[%s] for new sequence don't match to the prior offset[%s]\",\n              sequenceMetadata.getExclusiveStartPartitions(),\n              latestSequence\n          );\n        }\n      }\n","sourceCodeStart":1320,"sourceCodeEnd":1356,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java#L1320-L1356","documentation":"When a SeekableStream (Kafka/Kinesis) index task resumes from persisted checkpoints, getOrCreateSequenceMetadata verifies that each new sequence's start offsets exactly equal the end offsets recorded in the latest existing SequenceMetadata for each partition. If the new start offset differs from the prior end offset for a partition, it throws this IllegalStateException because the stream would be re-read from a wrong point (data loss/duplication).","triggerScenarios":"Task restart/resume where the checkpoints saved by a prior task generation don't line up: sequenceMetadata endOffsets for partition P differ from the startOffsets of the new sequence entry, typically after manually editing checkpoints, restoring from an older supervisor state, or a producer-consumer offset mismatch.","commonSituations":"Manually modified checkpoint file / task storage; resetting offsets via Kafka tooling while a task's sequence metadata still holds old offsets; supervisor resume after task failure where repartitioning changed partition IDs; restoring tasks from a different cluster.","solutions":["Reset the supervisor's offsets to match current stream state (POST /druid/indexer/v1/supervisor/<id>/reset) so checkpoints and sequence metadata are rebuilt.","Ensure you did not manually alter checkpoint JSON; restore consistent checkpoints where new-sequence startOffsets equal prior-sequence endOffsets.","If partitions were added/removed in the source topic, restart the supervisor so sequence metadata is regenerated for the new partition layout.","Verify task's persisted checkpoints in the tasks storage / database and correct them to align endOffsets with the new start offsets."],"exampleFix":"// before: new sequence startOffsets don't match prior endOffsets\n{\"startOffsets\": {\"0\": 100}} // prior endOffsets: {\"0\": 250}\n// after: reset supervisor to rebuild checkpoints\ncurl -X POST 'http://overlord:8087/druid/indexer/v1/supervisor/my-supervisor/reset'","handlingStrategy":"validation","validationCode":"// before resuming, verify checkpoints are contiguous\n// new sequence startOffsets must equal prior sequence endOffsets per partition\nif (!newSeq.getStartOffsets().equals(latestSeq.getEndOffsets())) { resetSupervisor(); }","typeGuard":null,"tryCatchPattern":"try { resumeSupervisor(id); } catch (IllegalStateException e) { if (e.getMessage().contains(\"does not equal expected prior offset\")) { resetSupervisor(id); } else { throw e; } }","preventionTips":["Never hand-edit checkpoint metadata","Use the supervisor reset endpoint after offset manipulation","Avoid changing topic offsets while tasks are mid-sequence"],"tags":["druid","kafka","kinesis","checkpoints","offsets"],"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-14T05:17:10.506Z"}