Lag metric: rabbit partitions

Error message

Lag metric: rabbit partitions %s do not match task partitions %s

What it means

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.

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

Defensive patterns

Strategy: validation

Validate before calling

Set<String> streamPartitions = latestSequenceFromStream.keySet();
Set<String> taskPartitions = highestCurrentOffsets.keySet();
if (!streamPartitions.equals(taskPartitions)) {
  // roll supervisor / resubscribe before trusting lag metrics
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/0b49172a9586d70b. Report an issue: GitHub.

Appendix: source

Thrown at extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java:255

          context,
          sortingMapper,
          CollectionUtils.isNullOrEmpty(serverPrioritiesToAssign) ? null : serverPrioritiesToAssign.get(i)
      ));
    }
    return taskList;
  }

  @Override
  protected Map<String, Long> getPartitionRecordLag()
  {
    Map<String, Long> highestCurrentOffsets = getHighestCurrentOffsets();

    if (latestSequenceFromStream == null) {
      return null;
    }

    if (!latestSequenceFromStream.keySet().equals(highestCurrentOffsets.keySet())) {
      log.warn(
          "Lag metric: rabbit partitions %s do not match task partitions %s",
          latestSequenceFromStream.keySet(),
          highestCurrentOffsets.keySet());
    }

    return getRecordLagPerPartitionInLatestSequences(highestCurrentOffsets);
  }

  @Nullable
  @Override
  protected Map<String, Long> getPartitionTimeLag()
  {
    // time lag not currently support with rabbit
    return null;
  }

  // suppress use of CollectionUtils.mapValues() since the valueMapper function
  // is

View on GitHub (pinned to 9b90983fd2)