{"record":{"id":"7c5174e0d639f0d3","repo":"apache/druid","slug":"driver-for-sequence-s-attempted-to-publish-inval","errorCode":null,"errorMessage":"Driver for sequence[%s] attempted to publish invalid metadata[%s].","messagePattern":"Driver for sequence\\[(.+?)\\] attempted to publish invalid metadata\\[(.+?)\\]\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"critical","filePath":"indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SequenceMetadata.java","lineNumber":374,"sourceCode":"        SegmentSchemaMapping segmentSchemaMapping\n    ) throws IOException\n    {\n      if (mustBeNullOrEmptyOverwriteSegments != null && !mustBeNullOrEmptyOverwriteSegments.isEmpty()) {\n        throw new ISE(\n            \"Stream ingestion task unexpectedly attempted to overwrite segments: %s\",\n            SegmentUtils.commaSeparatedIdentifiers(mustBeNullOrEmptyOverwriteSegments)\n        );\n      }\n      final Map<?, ?> commitMetaMap = (Map<?, ?>) Preconditions.checkNotNull(commitMetadata, \"commitMetadata\");\n      final SeekableStreamEndSequenceNumbers<PartitionIdType, SequenceOffsetType> finalPartitions =\n          runner.deserializePartitionsFromMetadata(\n              toolbox.getJsonMapper(),\n              commitMetaMap.get(SeekableStreamIndexTaskRunner.METADATA_PUBLISH_PARTITIONS)\n          );\n\n      // Sanity check, we should only be publishing things that match our desired end state.\n      if (!getEndOffsets().equals(finalPartitions.getPartitionSequenceNumberMap())) {\n        throw new ISE(\n            \"Driver for sequence[%s] attempted to publish invalid metadata[%s].\",\n            SequenceMetadata.this.toString(),\n            commitMetadata\n        );\n      }\n\n      final TaskAction<SegmentPublishResult> action;\n\n      if (segmentsToPush.isEmpty()) {\n        // If a task ingested no data but made progress reading through its assigned partitions,\n        // we publish no segments but still need to update the supervisor with the current offsets\n        SeekableStreamSequenceNumbers<PartitionIdType, SequenceOffsetType> startPartitions =\n            new SeekableStreamStartSequenceNumbers<>(\n                finalPartitions.getStream(),\n                getStartOffsets(),\n                exclusiveStartPartitions\n            );\n        if (isMetadataUnchanged(startPartitions, finalPartitions)) {","sourceCodeStart":356,"sourceCodeEnd":392,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SequenceMetadata.java#L356-L392","documentation":"During segment publishing, SeekableStreamIndexTaskRunner hands SequenceMetadata the commit metadata (partition->sequence offsets) it stored. Before publishing segments, SequenceMetadata verifies that the final partitions implied by the stored commit metadata exactly match this SequenceMetadata's endOffsets. A mismatch means the task's in-memory publish state and its persisted commit metadata disagree, so publishing is aborted with an IllegalStateException to prevent committing wrong offsets.","triggerScenarios":"publishAnnotatedSegments is called with a commitMetaMap whose METADATA_PUBLISH_PARTITIONS entry does not equal getEndOffsets() of this SequenceMetadata — e.g. publishing after a checkpoint changed the sequence's end offsets, or replaying/storing stale commit metadata from a previous sequence.","commonSituations":"Kafka/Kinesis tasks crashing and resuming with persisted commit metadata from an older checkpoint; replica tasks with divergent checkpoints; bugs or manual edits to task storage that leave stale metadata rows behind.","solutions":["Compare the metadata[%s] value in the message with the task's expected end offsets; if stale, reset the supervisor/sequence so tasks restart from consistent offsets","Kill and re-launch the affected tasks so a fresh SequenceMetadata and commit metadata are generated","Check whether a checkpoint request (supervisor checkpoint API) was issued with offsets inconsistent with the running sequence and re-checkpoint correctly","If reproducible, verify no duplicate publishing path stores commit metadata for the wrong sequence; upgrade Druid if a known bug"],"exampleFix":"// before: publishing with mismatched commit metadata\nrunner.publishAnnotatedSegments(toolbox, staleCommitMetaMap);\n// after: verify sequence matches end offsets before publishing\nif (!sequenceMetadata.getEndOffsets().equals(commitMetaMap.get(METADATA_PUBLISH_PARTITIONS))) {\n  // checkpoint/reset or skip publishing for this stale metadata\n}","handlingStrategy":"validation","validationCode":"if (!sequenceMetadata.getEndOffsets().equals(commitMeta.getPartitionSequenceNumberMap())) {\n  throw new IllegalStateException(\"Commit metadata does not match current sequence end offsets; reset or re-checkpoint before publishing\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  sequenceMetadata.publishAnnotatedSegments(toolbox, commitMetaMap);\n} catch (IllegalStateException e) {\n  log.error(e, \"Invalid publish metadata; aborting publish and resetting sequence\");\n  // trigger supervisor reset / task restart\n}","preventionTips":["Always checkpoint through the supervisor API so commit metadata stays consistent with sequences","Avoid manually editing task/segment metadata in the metadata store","Monitor task logs for checkpoint mismatches before they reach publish"],"tags":["kafka","kinesis","ingestion","state-consistency"],"backgroundTag":"internal-invariant-violation","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"}