{"record":{"id":"1ab336cb9f353865","repo":"apache/druid","slug":"could-not-fetch-partitions-for-topic-stream-s-1ab336","errorCode":null,"errorMessage":"Could not fetch partitions for topic/stream [%s]","messagePattern":"Could not fetch partitions for topic/stream \\[(.+?)\\]","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":459,"sourceCode":"  }\n\n  /**\n   * This method is similar to updatePartitionLagFromStream\n   * but also determines time lag. Once this method has been\n   * tested, we can remove the older one.\n   */\n  private void updatePartitionTimeAndRecordLagFromStream()\n  {\n    final Map<KafkaTopicPartition, Long> highestCurrentOffsets = getHighestCurrentOffsets();\n\n    getRecordSupplierLock().lock();\n    try {\n      Set<KafkaTopicPartition> partitionIds;\n      try {\n        partitionIds = recordSupplier.getPartitionIds(getIoConfig().getStream());\n      }\n      catch (Exception e) {\n        log.warn(\"Could not fetch partitions for topic/stream [%s]\", getIoConfig().getStream());\n        throw new StreamException(e);\n      }\n\n      final Set<StreamPartition<KafkaTopicPartition>> partitions = partitionIds\n          .stream()\n          .map(e -> new StreamPartition<>(getIoConfig().getStream(), e))\n          .collect(Collectors.toSet());\n\n      // Since we cannot compute the current timestamp for partitions for\n      // which we haven't started reading yet explictly set them.\n      final Set<KafkaTopicPartition> yetToReadPartitions = new HashSet<>();\n      for (KafkaTopicPartition partition : partitionIds) {\n        Long highestCurrentOffset = highestCurrentOffsets.get(partition);\n        if (highestCurrentOffset == null || highestCurrentOffset == 0) {\n          yetToReadPartitions.add(partition);\n        } else {\n          recordSupplier.seek(new StreamPartition<>(getIoConfig().getStream(), partition), highestCurrentOffset - 1);\n        }","sourceCodeStart":441,"sourceCodeEnd":477,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java#L441-L477","documentation":"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.","triggerScenarios":"recordSupplier.getPartitionIds(getIoConfig().getStream()) throwing any Exception — e.g. Kafka broker unreachable, authorization failure, or unknown topic — during updatePartitionTimeAndRecordLagFromStream.","commonSituations":"Kafka cluster outage or network partition; wrong bootstrap servers in spec; topic renamed or deleted while supervisor runs; ACLs denying metadata requests.","solutions":["Fix connectivity to Kafka: verify bootstrap.servers, DNS, and firewall rules in the supervisor spec.","Confirm the topic/stream name in the spec exists in the cluster.","Check Kafka ACLs allow DESCRIBE on the topic and CLUSTER metadata.","Inspect the wrapped StreamException cause for the underlying client error."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Pre-check reachability before submitting/monitoring the supervisor\ntry (KafkaConsumer c = new KafkaConsumer(props)) { c.partitionsFor(topic); }","typeGuard":null,"tryCatchPattern":"try { lag = supervisorLag(); } catch (StreamException e) { log.error(\"partition fetch failed\", e.getCause()); } // handle wrapped cause","preventionTips":["Validate bootstrap.servers and topic existence at spec submission","Grant Kafka DESCRIBE ACLs to the supervisor identity","Alert on repeated StreamException in supervisor logs"],"tags":["kafka","partitions","supervisor","stream"],"backgroundTag":"http-error-response","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"}