xai-org/x-algorithm · error · ValueError

Region {region!r} must list at least one broker

Error message

Region {region!r} must list at least one broker

What it means

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.

Source

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

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())
        results = await asyncio.gather(
            *[self._start_region(region, brokers) for region, brokers in clusters],
            return_exceptions=True,

View on GitHub (pinned to 24c60942c5)

Solutions

1. Remove the empty region entry or give it real broker addresses.
2. If brokers come from a discovery step, make it fail loudly upstream instead of emitting an empty list.
3. Normalize config parsing: filter out empty regions before constructing the pydantic model and log a warning.

Example fix

# before
cfg = MultiRegionKafkaConsumerConfig(clusters={'us': ['b:9092'], 'eu': []})  # ValueError: Region 'eu'

# after
cfg = MultiRegionKafkaConsumerConfig(clusters={'us': ['b:9092'], 'eu': ['eu-b1:9092']})

clusters = {r: b for r, b in raw_clusters.items() if b}
if not clusters:
    raise SystemExit('no kafka regions with brokers configured')
cfg = MultiRegionKafkaConsumerConfig(clusters=clusters)

try:
    cfg = MultiRegionKafkaConsumerConfig(clusters=raw)
except ValueError as e:
    # message names the offending region
    logger.error('cluster config invalid: %s', e)
    raise

When it happens

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

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

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

• `clusters` must contain at least one region
• `clusters` must contain at least one region
• Region {region!r} must list at least one broker
• acks must be 0, 1, or 'all', got {self.acks!r}
• invalid seekStrage ${config.seekStrategy}

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