xai-org/x-algorithm · error · RuntimeError

Producer not started

Error message

Producer not started

What it means

MultiRegionKafkaProducer.send() requires that start() has populated self._producers; sending before start (or after stop/when no producers were successfully started) raises RuntimeError. It is a lifecycle guard analogous to the consumer's poll guard.

Source

Thrown at grox/libs/kafka_cli/multi_region_producer.py:131

                logger.exception("Producer region retry task failed during shutdown")
            self._region_retry_task = None
        for region, producer in self._producers.items():
            try:
                await producer.stop()
            except Exception:
                logger.exception(f"Failed to stop producer for region {region!r}")
        self._producers = {}

    def _candidate_regions(self) -> list[str]:
        candidates = [
            region for region in self.config.clusters if region in self._producers
        ]
        random.shuffle(candidates)
        return candidates

    async def send(self, id: str, value: bytes):
        if not self._producers:
            raise RuntimeError("Producer not started")
        candidates = self._candidate_regions()
        if not candidates:
            raise RuntimeError(
                f"No healthy Kafka region available for topic {self.topic!r}"
            )
        start = time.perf_counter()
        errors: list[BaseException] = []
        for region in candidates:
            attributes = {"topic": self.topic, "region": region}
            try:
                await self._producers[region].send_and_wait(
                    self.topic, key=id.encode(), value=value
                )
                Metrics.counter("kafka_producer.sent.count").add(
                    1, attributes=attributes
                )
                Metrics.histogram("kafka_producer.send_duration").record(
                    time.perf_counter() - start, attributes=attributes

View on GitHub (pinned to 24c60942c5)

Solutions

1. Call and await start() during application startup (FastAPI lifespan, service init) before any send path is reachable.
2. Delay or queue sends until the producer reports started; check an is_started flag if exposed.
3. Ensure shutdown drains/cancels producer tasks before stop().

Example fix

# before
producer = MultiRegionKafkaProducer(cfg)
await producer.send('k1', b'v1')  # RuntimeError: Producer not started

# after
producer = MultiRegionKafkaProducer(cfg)
await producer.start()
await producer.send('k1', b'v1')

await producer.start()
assert producer._producers, 'producer failed to start'

try:
    await producer.send(k, v)
except RuntimeError as e:
    if 'not started' in str(e):
        await producer.start()
        await producer.send(k, v)
    else:
        raise

When it happens

Trigger: awaiting producer.send(id, value) before awaiting producer.start(), or after stop() has torn down the per-region producers.

Common situations: Module-level producer instance used by request handlers before the app's startup hook ran; a background task emitting metrics after shutdown began; start() partially failed leaving _producers empty.

Related errors

• Consumer not started
• Task generators not started
• `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}

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