xai-org/x-algorithm · error · ValueError

`clusters` must contain at least one region

Error message

`clusters` must contain at least one region

What it means

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.

Source

Thrown at grox/libs/kafka_cli/multi_region_consumer.py:30

from pydantic import BaseModel, Field, model_validator

logger = logging.getLogger(__name__)


class MultiRegionKafkaConsumerConfig(BaseModel):
    topic: str
    group_id: str
    clusters: dict[str, list[str]]
    auto_offset_reset: str = Field(default="latest")
    fetch_max_bytes: int = Field(default=50 * 1024 * 1024)
    fetch_min_bytes: int = Field(default=1024 * 128)
    max_poll_records: int = Field(default=500)
    request_timeout_ms: int = Field(default=30000)

    @model_validator(mode="after")
    def _validate_clusters(self) -> "MultiRegionKafkaConsumerConfig":
        if not self.clusters:
            raise ValueError("`clusters` must contain at least one region")
        for region, brokers in self.clusters.items():
            if not brokers:
                raise ValueError(f"Region {region!r} must list at least one broker")
        return self


RETRY_FAILED_REGION_INTERVAL_SEC = 60


class MultiRegionKafkaConsumer:
    def __init__(self, config: MultiRegionKafkaConsumerConfig):
        self.config = config
        self.group_id: str = config.group_id
        self._consumers: dict[str, AIOKafkaConsumer] = {}
        self._region_retry_task: asyncio.Task | None = None

    async def start(self):
        clusters = list(self.config.clusters.items())

View on GitHub (pinned to 24c60942c5)

Solutions

  1. Populate clusters with at least one region, e.g. {'us-east': ['broker1:9092']}.
  2. Fix the upstream config source (YAML/JSON/env template) so the clusters key is present and non-empty.
  3. Add a startup smoke test that validates the parsed config object before the app boots.

Example fix

# before
cfg = MultiRegionKafkaConsumerConfig(clusters={})  # ValueError

# after
cfg = MultiRegionKafkaConsumerConfig(
    clusters={'us-east-1': ['kafka-0.us-east:9092', 'kafka-1.us-east:9092']}
)
Defensive patterns

Strategy: validation

Validate before calling

if not clusters:
    raise SystemExit('kafka clusters config is empty; check config source')
cfg = MultiRegionKafkaConsumerConfig(clusters=clusters)

Try / catch

try:
    cfg = MultiRegionKafkaConsumerConfig(**raw)
except ValueError as e:
    logger.error('invalid consumer config: %s', e)
    raise

Prevention

When it happens

Trigger: 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 {}.

Common situations: 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.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/7cf08ae8f854b80c. Report an issue: GitHub.