apache/druid · warning

Could not fetch partitions for topic/stream

Error message

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

What it means

In KafkaSupervisor.updatePartitionTimeAndRecordLagFromStream(), when recordSupplier.getPartitionIds() throws, the supervisor logs 'Could not fetch partitions for topic/stream [%s]' and wraps the cause in a StreamException to propagate it to the caller (updatePartitionLagFromStream). It indicates the supervisor could not enumerate topic partitions while computing lag.

Solutions

1. Fix connectivity to Kafka: verify bootstrap.servers, DNS, and firewall rules in the supervisor spec.
2. Confirm the topic/stream name in the spec exists in the cluster.
3. Check Kafka ACLs allow DESCRIBE on the topic and CLUSTER metadata.
4. Inspect the wrapped StreamException cause for the underlying client error.

// Pre-check reachability before submitting/monitoring the supervisor
try (KafkaConsumer c = new KafkaConsumer(props)) { c.partitionsFor(topic); }

try { lag = supervisorLag(); } catch (StreamException e) { log.error("partition fetch failed", e.getCause()); } // handle wrapped cause

When it happens

Trigger: recordSupplier.getPartitionIds(getIoConfig().getStream()) throwing any Exception — e.g. Kafka broker unreachable, authorization failure, or unknown topic — during updatePartitionTimeAndRecordLagFromStream.

Common situations: Kafka cluster outage or network partition; wrong bootstrap servers in spec; topic renamed or deleted while supervisor runs; ACLs denying metadata requests.

Understand the failure class

Background: "API error: {status}" and "HTTP 401/403/404/429/5xx" errors: non-2xx HTTP responses explained — this error's family across 27 libraries.

Related errors

• Mismatched kafka and task partitions: Missing Task…
• got null sequence number for partition
• Killing task[ ] as it failed to return start time.
• No such previous checkpoint
• No valid task counts after applying constraints for…

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

Appendix: source

Thrown at extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:459

  }

  /**
   * This method is similar to updatePartitionLagFromStream
   * but also determines time lag. Once this method has been
   * tested, we can remove the older one.
   */
  private void updatePartitionTimeAndRecordLagFromStream()
  {
    final Map<KafkaTopicPartition, Long> highestCurrentOffsets = getHighestCurrentOffsets();

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

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

      // Since we cannot compute the current timestamp for partitions for
      // which we haven't started reading yet explictly set them.
      final Set<KafkaTopicPartition> yetToReadPartitions = new HashSet<>();
      for (KafkaTopicPartition partition : partitionIds) {
        Long highestCurrentOffset = highestCurrentOffsets.get(partition);
        if (highestCurrentOffset == null || highestCurrentOffset == 0) {
          yetToReadPartitions.add(partition);
        } else {
          recordSupplier.seek(new StreamPartition<>(getIoConfig().getStream(), partition), highestCurrentOffset - 1);
        }

View on GitHub (pinned to 9b90983fd2)