{"record":{"id":"5fe776b2528a339a","repo":"xai-org/x-algorithm","slug":"region-region-r-must-list-at-least-one-broker","errorCode":null,"errorMessage":"Region {region!r} must list at least one broker","messagePattern":"Region (.+?) must list at least one broker","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"grox/libs/kafka_cli/multi_region_consumer.py","lineNumber":33,"sourceCode":"\n\nclass MultiRegionKafkaConsumerConfig(BaseModel):\n    topic: str\n    group_id: str\n    clusters: dict[str, list[str]]\n    auto_offset_reset: str = Field(default=\"latest\")\n    fetch_max_bytes: int = Field(default=50 * 1024 * 1024)\n    fetch_min_bytes: int = Field(default=1024 * 128)\n    max_poll_records: int = Field(default=500)\n    request_timeout_ms: int = Field(default=30000)\n\n    @model_validator(mode=\"after\")\n    def _validate_clusters(self) -> \"MultiRegionKafkaConsumerConfig\":\n        if not self.clusters:\n            raise ValueError(\"`clusters` must contain at least one region\")\n        for region, brokers in self.clusters.items():\n            if not brokers:\n                raise ValueError(f\"Region {region!r} must list at least one broker\")\n        return self\n\n\nRETRY_FAILED_REGION_INTERVAL_SEC = 60\n\n\nclass MultiRegionKafkaConsumer:\n    def __init__(self, config: MultiRegionKafkaConsumerConfig):\n        self.config = config\n        self.group_id: str = config.group_id\n        self._consumers: dict[str, AIOKafkaConsumer] = {}\n        self._region_retry_task: asyncio.Task | None = None\n\n    async def start(self):\n        clusters = list(self.config.clusters.items())\n        results = await asyncio.gather(\n            *[self._start_region(region, brokers) for region, brokers in clusters],\n            return_exceptions=True,","sourceCodeStart":15,"sourceCodeEnd":51,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/grox/libs/kafka_cli/multi_region_consumer.py#L15-L51","documentation":"The same pydantic _validate_clusters validator iterates clusters and requires every region key to map to a non-empty list of brokers. A region whose value is [] (or otherwise falsy) fails with a region-specific message, so you can identify which entry is broken.","triggerScenarios":"MultiRegionKafkaConsumerConfig with a region mapped to an empty list, e.g. {'us-east': ['b:9092'], 'eu-west': []}; commonly from dynamic broker discovery or templated config that yields zero brokers for one region.","commonSituations":"Config generated per-region where one environment has no brokers yet; a Helm range producing an empty list; env-parsed broker strings that split into nothing (empty string '').","solutions":["Remove the empty region entry or give it real broker addresses.","If brokers come from a discovery step, make it fail loudly upstream instead of emitting an empty list.","Normalize config parsing: filter out empty regions before constructing the pydantic model and log a warning."],"exampleFix":"# before\ncfg = MultiRegionKafkaConsumerConfig(clusters={'us': ['b:9092'], 'eu': []})  # ValueError: Region 'eu'\n\n# after\ncfg = MultiRegionKafkaConsumerConfig(clusters={'us': ['b:9092'], 'eu': ['eu-b1:9092']})","handlingStrategy":"validation","validationCode":"clusters = {r: b for r, b in raw_clusters.items() if b}\nif not clusters:\n    raise SystemExit('no kafka regions with brokers configured')\ncfg = MultiRegionKafkaConsumerConfig(clusters=clusters)","typeGuard":null,"tryCatchPattern":"try:\n    cfg = MultiRegionKafkaConsumerConfig(clusters=raw)\nexcept ValueError as e:\n    # message names the offending region\n    logger.error('cluster config invalid: %s', e)\n    raise","preventionTips":["Lint that every region value is a non-empty list of host:port","Fail broker discovery loudly instead of emitting empty lists","Alert when a region's broker count drops to zero"],"tags":["kafka","pydantic","config-validation","brokers"],"backgroundTag":"schema-validation-failed","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}