{"record":{"id":"f90f7712dc07d353","repo":"xai-org/x-algorithm","slug":"producer-not-started","errorCode":null,"errorMessage":"Producer not started","messagePattern":"Producer not started","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"grox/libs/kafka_cli/multi_region_producer.py","lineNumber":131,"sourceCode":"                logger.exception(\"Producer region retry task failed during shutdown\")\n            self._region_retry_task = None\n        for region, producer in self._producers.items():\n            try:\n                await producer.stop()\n            except Exception:\n                logger.exception(f\"Failed to stop producer for region {region!r}\")\n        self._producers = {}\n\n    def _candidate_regions(self) -> list[str]:\n        candidates = [\n            region for region in self.config.clusters if region in self._producers\n        ]\n        random.shuffle(candidates)\n        return candidates\n\n    async def send(self, id: str, value: bytes):\n        if not self._producers:\n            raise RuntimeError(\"Producer not started\")\n        candidates = self._candidate_regions()\n        if not candidates:\n            raise RuntimeError(\n                f\"No healthy Kafka region available for topic {self.topic!r}\"\n            )\n        start = time.perf_counter()\n        errors: list[BaseException] = []\n        for region in candidates:\n            attributes = {\"topic\": self.topic, \"region\": region}\n            try:\n                await self._producers[region].send_and_wait(\n                    self.topic, key=id.encode(), value=value\n                )\n                Metrics.counter(\"kafka_producer.sent.count\").add(\n                    1, attributes=attributes\n                )\n                Metrics.histogram(\"kafka_producer.send_duration\").record(\n                    time.perf_counter() - start, attributes=attributes","sourceCodeStart":113,"sourceCodeEnd":149,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/grox/libs/kafka_cli/multi_region_producer.py#L113-L149","documentation":"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.","triggerScenarios":"awaiting producer.send(id, value) before awaiting producer.start(), or after stop() has torn down the per-region producers.","commonSituations":"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.","solutions":["Call and await start() during application startup (FastAPI lifespan, service init) before any send path is reachable.","Delay or queue sends until the producer reports started; check an is_started flag if exposed.","Ensure shutdown drains/cancels producer tasks before stop()."],"exampleFix":"# before\nproducer = MultiRegionKafkaProducer(cfg)\nawait producer.send('k1', b'v1')  # RuntimeError: Producer not started\n\n# after\nproducer = MultiRegionKafkaProducer(cfg)\nawait producer.start()\nawait producer.send('k1', b'v1')","handlingStrategy":"validation","validationCode":"await producer.start()\nassert producer._producers, 'producer failed to start'","typeGuard":null,"tryCatchPattern":"try:\n    await producer.send(k, v)\nexcept RuntimeError as e:\n    if 'not started' in str(e):\n        await producer.start()\n        await producer.send(k, v)\n    else:\n        raise","preventionTips":["Initialize producers in app startup hooks before serving traffic","Buffer sends in an outbox until producer is ready","Drain producers on shutdown before stop()"],"tags":["kafka","producer","lifecycle","asyncio"],"backgroundTag":"resource-not-initialized","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}