{"record":{"id":"0b49172a9586d70b","repo":"apache/druid","slug":"lag-metric-rabbit-partitions-s-do-not-match-task","errorCode":null,"errorMessage":"Lag metric: rabbit partitions %s do not match task partitions %s","messagePattern":"Lag metric: rabbit partitions (.+?) do not match task partitions (.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java","lineNumber":255,"sourceCode":"          context,\n          sortingMapper,\n          CollectionUtils.isNullOrEmpty(serverPrioritiesToAssign) ? null : serverPrioritiesToAssign.get(i)\n      ));\n    }\n    return taskList;\n  }\n\n  @Override\n  protected Map<String, Long> getPartitionRecordLag()\n  {\n    Map<String, Long> highestCurrentOffsets = getHighestCurrentOffsets();\n\n    if (latestSequenceFromStream == null) {\n      return null;\n    }\n\n    if (!latestSequenceFromStream.keySet().equals(highestCurrentOffsets.keySet())) {\n      log.warn(\n          \"Lag metric: rabbit partitions %s do not match task partitions %s\",\n          latestSequenceFromStream.keySet(),\n          highestCurrentOffsets.keySet());\n    }\n\n    return getRecordLagPerPartitionInLatestSequences(highestCurrentOffsets);\n  }\n\n  @Nullable\n  @Override\n  protected Map<String, Long> getPartitionTimeLag()\n  {\n    // time lag not currently support with rabbit\n    return null;\n  }\n\n  // suppress use of CollectionUtils.mapValues() since the valueMapper function\n  // is","sourceCodeStart":237,"sourceCodeEnd":273,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java#L237-L273","documentation":"When computing lag for a Rabbit stream supervisor, the partition set seen in the latest stream sequences is compared to the partition set of current task offsets. If they differ (a partition appeared or disappeared, e.g. stream was resized), the lag calculation proceeds only on the task partitions and this warning is logged.","triggerScenarios":"latestSequenceFromStream.keySet() != highestCurrentOffsets.keySet(): the stream's partition count changed (stream trimmed/expanded via stream properties) while tasks still track old partitions, or recordSupplier returned a different partition set than the tasks report.","commonSituations":"RabbitMQ stream was re-created or its length/retention changed, adding/removing stream partitions; supervisor restarted with different partition count; a task lag query racing with supervisor partition rebalancing.","solutions":["Verify the stream wasn't resized/recreated; check RabbitMQ stream properties (max-length, initial cluster size)","Restart/roll the supervisor so tasks pick up the current partition set","Confirm getPartitionIds returns the expected partitions for the stream name","Ignore if transient during rebalance; investigate if persistent — lag values for mismatched partitions will be missing"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"Set<String> streamPartitions = latestSequenceFromStream.keySet();\nSet<String> taskPartitions = highestCurrentOffsets.keySet();\nif (!streamPartitions.equals(taskPartitions)) {\n  // roll supervisor / resubscribe before trusting lag metrics\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Don't resize/recreate streams while the supervisor runs","Rebalance tasks after any stream partition change","Treat the warning as a signal to restart the supervisor if persistent"],"tags":["rabbitmq","lag-metrics","partition-mismatch","supervisor"],"backgroundTag":"partition-mismatch","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"}