{"record":{"id":"0b7e3fa96e02bc5f","repo":"apache/druid","slug":"topic-stream-in-metadata-storage-s-doesn-t-matc","errorCode":null,"errorMessage":"Topic/stream in metadata storage [%s] doesn't match spec topic/stream [%s], ignoring stored sequences","messagePattern":"Topic/stream in metadata storage \\[(.+?)\\] doesn't match spec topic/stream \\[(.+?)\\], ignoring stored sequences","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java","lineNumber":667,"sourceCode":"      SeekableStreamSequenceNumbers<KafkaTopicPartition, Long> partitions = ((KafkaDataSourceMetadata) dataSourceMetadata)\n          .getSeekableStreamSequenceNumbers();\n      if (partitions != null && partitions.getPartitionSequenceNumberMap() != null) {\n        Map<KafkaTopicPartition, Long> partitionOffsets = new HashMap<>();\n        Set<String> topicMisMatchLogged = new HashSet<>();\n        partitions.getPartitionSequenceNumberMap().forEach((kafkaTopicPartition, value) -> {\n          final String matchValue;\n          // previous offsets are from multi-topic config\n          if (kafkaTopicPartition.topic().isPresent()) {\n            matchValue = kafkaTopicPartition.topic().get();\n          } else {\n            // previous offsets are from single topic config\n            matchValue = partitions.getStream();\n          }\n\n          KafkaTopicPartition matchingTopicPartition = getMatchingKafkaTopicPartition(kafkaTopicPartition, matchValue);\n\n          if (matchingTopicPartition == null && !topicMisMatchLogged.contains(matchValue)) {\n            log.warn(\n                \"Topic/stream in metadata storage [%s] doesn't match spec topic/stream [%s], ignoring stored sequences\",\n                matchValue,\n                getIoConfig().getStream()\n            );\n            topicMisMatchLogged.add(matchValue);\n          }\n          if (matchingTopicPartition != null) {\n            partitionOffsets.put(matchingTopicPartition, value);\n          }\n        });\n        return partitionOffsets;\n      }\n    }\n\n    return Collections.emptyMap();\n  }\n\n  @Nullable","sourceCodeStart":649,"sourceCodeEnd":685,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java#L649-L685","documentation":"In KafkaSupervisor.getOffsetsFromMetadataStorage(), sequences stored in metadata storage belong to a topic/stream that does not match the current spec's topic. The supervisor logs a warning (once per mismatched topic) and ignores those stored sequence numbers, effectively starting fresh for the spec's topic. This guards against reusing offsets from a different topic.","triggerScenarios":"The spec's topic/stream was changed (or metadata storage holds entries from a previous spec) so getMatchingKafkaTopicPartition returns null for the stored stream name.","commonSituations":"Editing an existing supervisor spec to point at a new topic while reusing the old datasource name; copied/renamed supervisor configs; stale metadata-store rows from an earlier spec version.","solutions":["Update the supervisor spec so the topic matches what metadata storage holds, or accept the reset.","If a topic change is intended, also reset the datasource (reset supervisor) and expect consumption from scratch or earliest.","Clean stale supervisor entries in the metadata storage table if they belong to deleted specs.","Use a unique datasource name for a new topic to avoid colliding stored offsets."],"exampleFix":"// before (spec edited in place)\n\"topic\": \"new-topic\" // metadata store still has old-topic offsets -> mismatch warning\n// after\nPOST /druid/indexer/supervisor/<id>/reset  // clear stale offsets for the datasource\nthen submit spec with \"topic\": \"new-topic\"","handlingStrategy":"validation","validationCode":"// Compare spec topic with stored supervisor state before submitting\nSupervisorMetaData md = fetchMetaDataStoreEntry(datasource);\nif (md != null && !Objects.equals(md.getSpec().getIoConfig().getTopic(), newSpec.getIoConfig().getTopic())) { resetSupervisor(datasource); }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Reset the supervisor when intentionally changing the topic","Use a distinct datasource name per topic","Clean up metadata-store rows of retired supervisors"],"tags":["kafka","metadata-storage","topic-mismatch","supervisor"],"backgroundTag":"conflicting-config-options","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"}