{"record":{"id":"905554ec30a76635","repo":"xai-org/x-algorithm","slug":"consumer-not-started","errorCode":null,"errorMessage":"Consumer not started","messagePattern":"Consumer not started","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"grox/libs/kafka_cli/multi_region_consumer.py","lineNumber":144,"sourceCode":"        if self._region_retry_task is not None:\n            self._region_retry_task.cancel()\n            try:\n                await self._region_retry_task\n            except asyncio.CancelledError:\n                pass\n            except Exception:\n                logger.exception(\"Region retry task failed during shutdown\")\n            self._region_retry_task = None\n        for region, consumer in self._consumers.items():\n            try:\n                await consumer.stop()\n            except Exception:\n                logger.exception(f\"Failed to stop consumer for region {region!r}\")\n        self._consumers = {}\n\n    async def poll(self, num: int) -> list[KafkaMessage]:\n        if not self._consumers:\n            raise RuntimeError(\"Consumer not started\")\n        Metrics.counter(\"kafka_consumer.fetching.count\").add(\n            num, attributes={\"group_id\": self.group_id}\n        )\n        start = time.perf_counter()\n        consumers = list(self._consumers.items())\n        max_records = max(1, num // len(consumers))\n        results = await asyncio.gather(\n            *[\n                consumer.getmany(timeout_ms=1000, max_records=max_records)\n                for _, consumer in consumers\n            ],\n            return_exceptions=True,\n        )\n        duration = time.perf_counter() - start\n        current_time = int(time.time())\n        Metrics.histogram(\"kafka_consumer.fetch_duration\").record(\n            duration, attributes={\"group_id\": self.group_id}\n        )","sourceCodeStart":126,"sourceCodeEnd":162,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/grox/libs/kafka_cli/multi_region_consumer.py#L126-L162","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Await start() (and confirm it succeeded) before entering the poll loop.","Guard the loop: if self._consumers is empty, re-start or exit the consuming task cleanly.","In shutdown paths, cancel/await polling tasks before calling stop()."],"exampleFix":"# before\nconsumer = MultiRegionKafkaConsumer(cfg)\nmsgs = await consumer.poll(100)  # RuntimeError: Consumer not started\n\n# after\nconsumer = MultiRegionKafkaConsumer(cfg)\nawait consumer.start()\nmsgs = await consumer.poll(100)","handlingStrategy":"validation","validationCode":"await consumer.start()\nassert consumer._consumers, 'consumer failed to start'","typeGuard":null,"tryCatchPattern":"try:\n    msgs = await consumer.poll(n)\nexcept RuntimeError as e:\n    if 'not started' in str(e):\n        await consumer.start()\n        msgs = await consumer.poll(n)\n    else:\n        raise","preventionTips":["Start consumers in app lifespan/startup hooks","Cancel poller tasks before stop() in shutdown","Guard poll loops with a started flag"],"tags":["kafka","lifecycle","asyncio","ordering"],"backgroundTag":"resource-not-initialized","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}