xai-org/x-algorithm · error · RuntimeError

Consumer not started

Error message

Consumer not started

What it means

MultiRegionKafkaConsumer.poll() requires that start() has already populated self._consumers; polling before start (or after stop(), which clears _consumers to {}) raises RuntimeError. It is a lifecycle/ordering guard, not a Kafka error.

Source

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

        if self._region_retry_task is not None:
            self._region_retry_task.cancel()
            try:
                await self._region_retry_task
            except asyncio.CancelledError:
                pass
            except Exception:
                logger.exception("Region retry task failed during shutdown")
            self._region_retry_task = None
        for region, consumer in self._consumers.items():
            try:
                await consumer.stop()
            except Exception:
                logger.exception(f"Failed to stop consumer for region {region!r}")
        self._consumers = {}

    async def poll(self, num: int) -> list[KafkaMessage]:
        if not self._consumers:
            raise RuntimeError("Consumer not started")
        Metrics.counter("kafka_consumer.fetching.count").add(
            num, attributes={"group_id": self.group_id}
        )
        start = time.perf_counter()
        consumers = list(self._consumers.items())
        max_records = max(1, num // len(consumers))
        results = await asyncio.gather(
            *[
                consumer.getmany(timeout_ms=1000, max_records=max_records)
                for _, consumer in consumers
            ],
            return_exceptions=True,
        )
        duration = time.perf_counter() - start
        current_time = int(time.time())
        Metrics.histogram("kafka_consumer.fetch_duration").record(
            duration, attributes={"group_id": self.group_id}
        )

View on GitHub (pinned to 24c60942c5)

Solutions

1. Await start() (and confirm it succeeded) before entering the poll loop.
2. Guard the loop: if self._consumers is empty, re-start or exit the consuming task cleanly.
3. In shutdown paths, cancel/await polling tasks before calling stop().

Example fix

# before
consumer = MultiRegionKafkaConsumer(cfg)
msgs = await consumer.poll(100)  # RuntimeError: Consumer not started

# after
consumer = MultiRegionKafkaConsumer(cfg)
await consumer.start()
msgs = await consumer.poll(100)

await consumer.start()
assert consumer._consumers, 'consumer failed to start'

try:
    msgs = await consumer.poll(n)
except RuntimeError as e:
    if 'not started' in str(e):
        await consumer.start()
        msgs = await consumer.poll(n)
    else:
        raise

When it happens

Trigger: Calling await consumer.poll(n) before awaiting consumer.start(); or calling poll after stop() has run (stop clears self._consumers = {} on the path shown); or a failed start leaving _consumers empty.

Common situations: Fast-path code or tests that skip the async start; shutdown races where a polling task outlives stop(); start() raising partway (some regions failed) so _consumers is empty when poll is attempted.

Related errors

• Producer not started
• Task generators not started
• Kafka mTLS requires CA at {ca_file!r} (mount internal-ca Con
• `clusters` must contain at least one region
• Region {region!r} must list at least one broker

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