{"record":{"id":"a6c8c6db62117993","repo":"xai-org/x-algorithm","slug":"clusters-must-contain-at-least-one-region-a6c8c6","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_producer.py","lineNumber":25,"sourceCode":"from kafka_cli.mtls import create_mtls_ssl_context\nfrom monitor.metrics import Metrics\nfrom pydantic import BaseModel, Field, model_validator\n\nlogger = logging.getLogger(__name__)\n\nRETRY_FAILED_REGION_INTERVAL_SEC = 60\n\n\nclass MultiRegionKafkaProducerConfig(BaseModel):\n    topic: str\n    clusters: dict[str, list[str]]\n    acks: int | str = Field(default=1)\n    request_timeout_ms: int = Field(default=30000)\n\n    @model_validator(mode=\"after\")\n    def _validate(self) -> \"MultiRegionKafkaProducerConfig\":\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        if self.acks not in (0, 1, \"all\"):\n            raise ValueError(f\"acks must be 0, 1, or 'all', got {self.acks!r}\")\n        return self\n\n\nclass MultiRegionKafkaProducer:\n    def __init__(self, config: MultiRegionKafkaProducerConfig):\n        self.config = config\n        self.topic: str = config.topic\n        self._producers: dict[str, AIOKafkaProducer] = {}\n        self._region_retry_task: asyncio.Task | None = None\n\n    async def start(self):\n        regions = list(self.config.clusters.items())\n        try:","sourceCodeStart":7,"sourceCodeEnd":43,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/grox/libs/kafka_cli/multi_region_producer.py#L7-L43","documentation":"MultiRegionKafkaProducerConfig mirrors the consumer config: a pydantic model_validator rejects an empty clusters mapping because the producer would have nowhere to send records. Construction of the config object itself fails.","triggerScenarios":"MultiRegionKafkaProducerConfig(clusters={}) or config loaded from a source where the producer's clusters section is missing/empty.","commonSituations":"Producer and consumer configs generated from the same template and the producer section was forgotten; deploying a producer-only service with a config intended for a different role; empty default after a schema migration.","solutions":["Provide at least one region with brokers in the producer config.","Fix the templating/config source that yields the empty mapping.","Validate the config at deploy time (e.g. in a helm hook or bootstrap check) so it fails before rollout."],"exampleFix":"# before\ncfg = MultiRegionKafkaProducerConfig(clusters={})  # ValueError\n\n# after\ncfg = MultiRegionKafkaProducerConfig(clusters={'us-east-1': ['kafka-0:9092']})","handlingStrategy":"validation","validationCode":"if not producer_clusters:\n    raise SystemExit('producer clusters config is empty')\ncfg = MultiRegionKafkaProducerConfig(clusters=producer_clusters)","typeGuard":null,"tryCatchPattern":"try:\n    cfg = MultiRegionKafkaProducerConfig(**raw)\nexcept ValueError as e:\n    logger.error('invalid producer config: %s', e)\n    raise","preventionTips":["Validate producer config in CI against a schema","Separate producer/consumer config templates so one section is never silently empty"],"tags":["kafka","pydantic","producer","config-validation"],"backgroundTag":"schema-validation-failed","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}