cocoindex-io/cocoindex · error · RuntimeError

Received an Iggy message for an untracked partition. Use an

Error message

Received an Iggy message for an untracked partition. Use an explicit partition_id per TopicStream instance.

What it means

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.

Source

Thrown at python/cocoindex/connectors/iggy/_source.py:205

        """Create and register a partition state."""
        state = _PartitionState(
            consumer=consumer,
            stream=stream,
            topic=topic,
            partition=partition,
            high_watermark=high_watermark,
            committed_next_offset=committed_next_offset,
            on_commit=self._check_ready,
        )
        self._partitions[partition] = state
        return state

    def get(self, partition: int) -> _PartitionState:
        """Get an initialized partition state."""
        try:
            return self._partitions[partition]
        except KeyError as e:
            raise RuntimeError(
                "Received an Iggy message for an untracked partition. "
                "Use an explicit partition_id per TopicStream instance."
            ) from e

    def mark_initialized(self) -> None:
        """Mark initial partition state loaded and check readiness."""
        self._initialized = True
        self._check_ready()

    def discard_all(self) -> None:
        """Discard all partition states."""
        for state in self._partitions.values():
            state.discard()
        self._partitions.clear()


def _committed_next_offset(stored_offset: int | None) -> int:
    """Convert Iggy's last-stored offset into the next offset to consume."""

View on GitHub (pinned to e84aa99b32)

Solutions

1. Create a separate TopicStream per partition, passing an explicit partition_id to each
2. Verify the topic's partition list and restrict consumption to the partition you configured
3. Recreate/reinitialize the consumer so all intended partitions are registered before polling

Example fix

// before
stream = iggy.TopicStream(client, stream="s", topic="t")
// after
stream = iggy.TopicStream(client, stream="s", topic="t", partition_id=0)

assert partition_id is not None, "Create one TopicStream per partition with an explicit partition_id"

def is_tracked(stream, partition: int) -> bool:
    return partition in stream._partitions

try:
    state = partitions.get(pid)
except RuntimeError as e:
    logger.error("Untracked partition %s; recreate TopicStream with explicit partition_id", pid)
    raise

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• The Python Iggy SDK does not expose per-partition high water
• Iggy topic {self._stream}/{self._topic} does not exist.
• coco.use_state() cannot be called inside a `with coco.compon
• coco.use_state() cannot be called inside a memoized function
• apache-iggy is required to use the Iggy connector. Please in

AI-assisted analysis of cocoindex-io/cocoindex@e84aa99b32 (2026-09-08). Data as JSON: /api/errors/cbc93fbccd44208a. Report an issue: GitHub.