apache/druid · error · StreamException

Could not fetch partitions for topic/stream

Error message

Could not fetch partitions for topic/stream [%s]

What it means

RabbitStreamSupervisor.updatePartitionLagFromStream could not obtain the partition IDs for the configured stream — the record supplier threw while querying RabbitMQ. The supervisor logs a warning, releases the record-supplier lock, and rethrows as a StreamException, since partition lag cannot be computed without partitions. Common causes are connectivity failures or a nonexistent/deleted stream.

Solutions

1. Inspect the wrapped exception for the root cause (connection refused, auth failure, stream not found).
2. Verify the RabbitMQ stream exists and the configured username/password/port are correct.
3. Check network reachability from the supervisor/overlord host to the RabbitMQ stream nodes.
4. After fixing connectivity, the supervisor can re-attempt lag computation on the next cycle.

When it happens

Trigger: Thrown at extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisor.java:422 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/0edd5db16cf0f881. 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:422

      return new LagStats(0, 0, 0);
    }

    return aggregatePartitionLags(partitionRecordLag);
  }

  @Override
  public void updatePartitionLagFromStream()
  {
    getRecordSupplierLock().lock();

    Set<String> partitionIds;
    try {
      partitionIds = recordSupplier.getPartitionIds(getIoConfig().getStream());
    }
    catch (Exception e) {
      log.warn("Could not fetch partitions for topic/stream [%s]", getIoConfig().getStream());
      getRecordSupplierLock().unlock();
      throw new StreamException(e);
    }

    Set<StreamPartition<String>> partitions = partitionIds
        .stream()
        .map(e -> new StreamPartition<>(getIoConfig().getStream(), e))
        .collect(Collectors.toSet());

    latestSequenceFromStream = partitions.stream()
        .collect(Collectors.toMap(StreamPartition::getPartitionId, recordSupplier::getLatestSequenceNumber));

    getRecordSupplierLock().unlock();

  }

  @Override
  public Map<String, Long> getLatestSequencesFromStream()
  {
    return latestSequenceFromStream != null ? latestSequenceFromStream : new HashMap<>();

View on GitHub (pinned to 9b90983fd2)