{"record":{"id":"32c40276f77d6d5b","repo":"xai-org/x-algorithm","slug":"no-partitions-found-for-topic-topic-available-t-32c402","errorCode":null,"errorMessage":"No partitions found for topic {topic}, available topics: {available}","messagePattern":"No partitions found for topic (.+?), available topics: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"critical","filePath":"phoenix/xrex/data/streaming/kafkaconsumer.py","lineNumber":991,"sourceCode":"        security_protocol=\"SASL_SSL\",\n        sasl_kerberos_domain_name=\"kafka\",\n        sasl_kerberos_service_name=\"kafka\",\n        sasl_mechanism=sasl_mechanism,\n        sasl_plain_username=sasl_plain_username,\n        sasl_plain_password=sasl_plain_password,\n        ssl_context=ssl_ctx,\n        request_timeout_ms=30000,\n    )\n    await discovery_consumer.start()\n    try:\n        all_partitions = discovery_consumer.partitions_for_topic(topic)\n        if not all_partitions:\n            all_partitions = await force_kafka_metadata_update(discovery_consumer, topic)\n        if not all_partitions:\n            all_partitions = await force_kafka_metadata_update(discovery_consumer, topic)\n        if not all_partitions:\n            available = await discovery_consumer.topics()\n            raise ValueError(\n                f\"No partitions found for topic {topic}, available topics: {available}\"\n            )\n    finally:\n        try:\n            await discovery_consumer.stop()\n        except asyncio.CancelledError:\n            pass\n\n    total_partitions = len(all_partitions)\n    partition_ids = _range_partitions(total_partitions, shard_index, num_shards)\n    shard_partitions = [TopicPartition(topic, i) for i in partition_ids]\n    if not partition_ids:\n        rank_logger.warning(\n            f\"Shard {shard_index} received 0 partitions (total={total_partitions}, \"\n            f\"num_shards={num_shards}).  This shard will be idle.\"\n        )\n    rank_logger.info(\n        f\"Discovered {total_partitions} partitions for {topic}, \"","sourceCodeStart":973,"sourceCodeEnd":1009,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkaconsumer.py#L973-L1009","documentation":"In the multi-consumer path, _consume_multi_consumer tries three times (initial partitions_for_topic plus two forced metadata updates) to discover the topic's partitions before fan-out; if all fail it raises with the list of topics the discovery consumer can see. It is the multi-consumer equivalent of the topic-not-found check.","triggerScenarios":"consume_messages entering the multi-consumer path for a topic that does not exist in cluster metadata; brokers unreachable so both forced refreshes return nothing; transient metadata propagation lag exceeding the two retries.","commonSituations":"Consumer starts before the topic is provisioned; wrong bootstrap_servers/environment; topic recently deleted and recreated during a migration; heavy cluster load delaying metadata propagation.","solutions":["Check the topic name against the 'available topics' in the message and fix config/environment mismatch.","Add a readiness wait: poll partitions_for_topic with backoff for 30-60s before calling consume_messages.","Pre-create topics via infra automation so consumers never race topic creation.","If persistent, verify broker connectivity and ACLs for the client principal."],"exampleFix":"# before\nawait consume_messages(topic=\"new-topic\", ...)\n\n# after\n# wait for topic to appear in metadata\nasync def wait_for_topic(consumer, topic, timeout=60):\n    deadline = asyncio.get_event_loop().time() + timeout\n    while asyncio.get_event_loop().time() < deadline:\n        if await consumer.partitions_for_topic(topic):\n            return\n        await asyncio.sleep(2)\n    raise TimeoutError(topic)\nawait wait_for_topic(consumer, \"new-topic\")\nawait consume_messages(topic=\"new-topic\", ...)","handlingStrategy":"retry","validationCode":"parts = await consumer.partitions_for_topic(topic)\nif not parts:\n    for _ in range(10):\n        await asyncio.sleep(2)\n        parts = await consumer.partitions_for_topic(topic)\n        if parts:\n            break","typeGuard":null,"tryCatchPattern":"try:\n    await consume_messages(...)\nexcept ValueError as e:\n    if \"No partitions found\" in str(e):\n        rank_logger.error(f\"Topic unavailable: {e}; retrying after backoff\")\n        await asyncio.sleep(30)\n        raise","preventionTips":["Deploy topics before consumers (infra ordering).","Add readiness probes/wait-for-topic in job startup scripts.","Retry with backoff around transient metadata emptiness."],"tags":["kafka","topic-not-found","metadata","multi-consumer"],"backgroundTag":"kafka-topic-not-found","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}