{"record":{"id":"8908ab02f798e5c8","repo":"xai-org/x-algorithm","slug":"no-partitions-found-for-topic-topic-available-t","errorCode":null,"errorMessage":"No partitions found for topic {topic}, available topics: {available_topics}","messagePattern":"No partitions found for topic (.+?), available topics: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"critical","filePath":"phoenix/xrex/data/streaming/kafkaconsumer.py","lineNumber":510,"sourceCode":"    )\n    assert not (seek_to_timestamp_ms is not None and reset_to_latest), (\n        \"seek_to_timestamp_ms and reset_to_latest cannot both be set\"\n    )\n    rank_logger.info(\n        f\"Try to get partition info for {topic}, shard_index: {shard_index}, num_shards: {num_shards}.\"\n    )\n\n    partitions: set[int] | None = consumer.partitions_for_topic(topic)\n    if not partitions:\n        partitions = await force_kafka_metadata_update(consumer, topic)\n    if not partitions:\n        partitions = await force_kafka_metadata_update(consumer, topic)\n    if not partitions:\n        available_topics = await consumer.topics()\n        rank_logger.error(\n            f\"No partitions found for topic {topic}, available topics: {available_topics}\"\n        )\n        raise ValueError(\n            f\"No partitions found for topic {topic}, available topics: {available_topics}\"\n        )\n\n    total_partitions = len(partitions)\n\n    partition_ids = _range_partitions(total_partitions, shard_index, num_shards)\n    if not partition_ids:\n        rank_logger.warning(\n            f\"Worker {shard_index} received 0 partitions (total_partitions={total_partitions}, \"\n            f\"num_shards={num_shards}).  This worker will be idle.\"\n        )\n    assigned_partitions = [TopicPartition(topic, i) for i in partition_ids]\n    for tp in assigned_partitions:\n        rank_logger.info(\n            f\"Assigned partition {tp} to worker {shard_index}, num_shards: {num_shards}, total_partitions: {total_partitions}.\"\n        )\n    consumer.assign(assigned_partitions)\n    rank_logger.info(","sourceCodeStart":492,"sourceCodeEnd":528,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkaconsumer.py#L492-L528","documentation":"Raised in handle_topic_offset when, after partitions_for_topic plus two forced metadata refreshes, no partitions for the requested topic are known; the error lists topics the consumer can see. This blocks the consume path because there is nothing to assign or seek.","triggerScenarios":"Calling consume_messages with a topic absent from the cluster metadata; brokers returning empty metadata because they are unreachable; topic created moments ago and metadata not yet propagated even after forced refresh.","commonSituations":"Race condition where the consumer starts before the topic is provisioned; wrong bootstrap servers for the environment; broker DNS resolving but brokers not accepting connections so topics() returns partial/empty results; ACL restrictions hiding the topic.","solutions":["Use the 'available topics' list in the message to confirm the topic name and cluster, fix config typos or environment mismatch.","Pre-create the topic and wait for metadata propagation (e.g. kafka-topics --describe) before starting the consumer.","Add a startup readiness check/retry loop that polls partitions_for_topic until non-empty or a timeout.","Verify network/ACL access to all brokers, not just the bootstrap one."],"exampleFix":"# before\nawait consume_messages(topic=\"events-prod\", ...)\n\n# after\nfor _ in range(30):\n    if await consumer.partitions_for_topic(topic):\n        break\n    await asyncio.sleep(2)\nawait consume_messages(topic=\"events-prod\", ...)  # proceed only when partitions visible","handlingStrategy":"retry","validationCode":"async def wait_for_partitions(consumer, topic, timeout=60.0) -> list:\n    deadline = time.monotonic() + timeout\n    while time.monotonic() < deadline:\n        parts = await consumer.partitions_for_topic(topic)\n        if parts:\n            return parts\n        await asyncio.sleep(2)\n    raise TimeoutError(f\"topic {topic} never became visible\")","typeGuard":null,"tryCatchPattern":"try:\n    await handle_topic_offset(...)\nexcept ValueError as e:\n    if \"available topics\" in str(e):\n        # log available list, fix config or retry after provisioning\n        ...","preventionTips":["Provision topics before deploying consumers.","Add readiness gates/wait loops for topic metadata.","Pin environment-specific bootstrap servers via config, not defaults."],"tags":["kafka","topic-not-found","metadata","retry"],"backgroundTag":"kafka-topic-not-found","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}