{"record":{"id":"cbc93fbccd44208a","repo":"cocoindex-io/cocoindex","slug":"received-an-iggy-message-for-an-untracked-partitio","errorCode":null,"errorMessage":"Received an Iggy message for an untracked partition. Use an explicit partition_id per TopicStream instance.","messagePattern":"Received an Iggy message for an untracked partition\\. Use an explicit partition_id per TopicStream instance\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"python/cocoindex/connectors/iggy/_source.py","lineNumber":205,"sourceCode":"        \"\"\"Create and register a partition state.\"\"\"\n        state = _PartitionState(\n            consumer=consumer,\n            stream=stream,\n            topic=topic,\n            partition=partition,\n            high_watermark=high_watermark,\n            committed_next_offset=committed_next_offset,\n            on_commit=self._check_ready,\n        )\n        self._partitions[partition] = state\n        return state\n\n    def get(self, partition: int) -> _PartitionState:\n        \"\"\"Get an initialized partition state.\"\"\"\n        try:\n            return self._partitions[partition]\n        except KeyError as e:\n            raise RuntimeError(\n                \"Received an Iggy message for an untracked partition. \"\n                \"Use an explicit partition_id per TopicStream instance.\"\n            ) from e\n\n    def mark_initialized(self) -> None:\n        \"\"\"Mark initial partition state loaded and check readiness.\"\"\"\n        self._initialized = True\n        self._check_ready()\n\n    def discard_all(self) -> None:\n        \"\"\"Discard all partition states.\"\"\"\n        for state in self._partitions.values():\n            state.discard()\n        self._partitions.clear()\n\n\ndef _committed_next_offset(stored_offset: int | None) -> int:\n    \"\"\"Convert Iggy's last-stored offset into the next offset to consume.\"\"\"","sourceCodeStart":187,"sourceCodeEnd":223,"githubUrl":"https://github.com/cocoindex-io/cocoindex/blob/e84aa99b3292c5270a4b313b2a7137ad9ce8ab3b/python/cocoindex/connectors/iggy/_source.py#L187-L223","documentation":"The Iggy consumer tracks state only for partitions it was explicitly initialized with. When a message arrives for a partition id not present in the internal _partitions map, get() converts the KeyError into a RuntimeError advising the use of an explicit partition_id per TopicStream instance, since the SDK cannot safely multiplex unknown partitions.","triggerScenarios":"Consuming a multi-partition Iggy topic with a TopicStream created without an explicit partition_id (or polling a partition different from the configured one), so a polled message's partition has no tracked state.","commonSituations":"Creating one TopicStream for a topic that later gets extra partitions; sharing a consumer across partitions; relying on auto-assignment when the server assigns partitions not tracked client-side.","solutions":["Create a separate TopicStream per partition, passing an explicit partition_id to each","Verify the topic's partition list and restrict consumption to the partition you configured","Recreate/reinitialize the consumer so all intended partitions are registered before polling"],"exampleFix":"// before\nstream = iggy.TopicStream(client, stream=\"s\", topic=\"t\")\n// after\nstream = iggy.TopicStream(client, stream=\"s\", topic=\"t\", partition_id=0)","handlingStrategy":"validation","validationCode":"assert partition_id is not None, \"Create one TopicStream per partition with an explicit partition_id\"","typeGuard":"def is_tracked(stream, partition: int) -> bool:\n    return partition in stream._partitions","tryCatchPattern":"try:\n    state = partitions.get(pid)\nexcept RuntimeError as e:\n    logger.error(\"Untracked partition %s; recreate TopicStream with explicit partition_id\", pid)\n    raise","preventionTips":["Always pass explicit partition_id when consuming multi-partition topics","Check topic partitions_count before starting consumers","Recreate streams if the topic's partition layout changed"],"tags":["iggy","partition","streaming","state"],"backgroundTag":"invalid-state-transition","analyzedSha":"e84aa99b3292c5270a4b313b2a7137ad9ce8ab3b","analyzedAt":"2026-09-08T15:59:19.997Z","contentChangedAt":"2026-09-08T15:59:19.997Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}