{"record":{"id":"7cf08ae8f854b80c","repo":"xai-org/x-algorithm","slug":"clusters-must-contain-at-least-one-region","errorCode":null,"errorMessage":"`clusters` must contain at least one region","messagePattern":"`clusters` must contain at least one region","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"grox/libs/kafka_cli/multi_region_consumer.py","lineNumber":30,"sourceCode":"from pydantic import BaseModel, Field, model_validator\n\nlogger = logging.getLogger(__name__)\n\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())","sourceCodeStart":12,"sourceCodeEnd":48,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/grox/libs/kafka_cli/multi_region_consumer.py#L12-L48","documentation":"MultiRegionKafkaConsumerConfig runs a pydantic model_validator (mode='after') that rejects a config with an empty clusters mapping. clusters is the dict of region->brokers that the multi-region consumer fans out to, so an empty dict means there is nothing to consume from and construction fails immediately.","triggerScenarios":"Constructing MultiRegionKafkaConsumerConfig(clusters={}) or omitting clusters entirely when its default is empty; also loading config from YAML/JSON where the clusters key is missing or set to {}.","commonSituations":"Environment-specific config file that forgot the kafka.clusters section; templating (Helm/jinja) rendering an empty mapping for a non-prod environment; passing clusters=None instead of a populated dict.","solutions":["Populate clusters with at least one region, e.g. {'us-east': ['broker1:9092']}.","Fix the upstream config source (YAML/JSON/env template) so the clusters key is present and non-empty.","Add a startup smoke test that validates the parsed config object before the app boots."],"exampleFix":"# before\ncfg = MultiRegionKafkaConsumerConfig(clusters={})  # ValueError\n\n# after\ncfg = MultiRegionKafkaConsumerConfig(\n    clusters={'us-east-1': ['kafka-0.us-east:9092', 'kafka-1.us-east:9092']}\n)","handlingStrategy":"validation","validationCode":"if not clusters:\n    raise SystemExit('kafka clusters config is empty; check config source')\ncfg = MultiRegionKafkaConsumerConfig(clusters=clusters)","typeGuard":null,"tryCatchPattern":"try:\n    cfg = MultiRegionKafkaConsumerConfig(**raw)\nexcept ValueError as e:\n    logger.error('invalid consumer config: %s', e)\n    raise","preventionTips":["Schema-validate config files at deploy time","Default clusters to a required field, not optional","Log the parsed clusters dict at startup"],"tags":["kafka","pydantic","config-validation","multi-region"],"backgroundTag":"schema-validation-failed","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}