Apache Iggy connector

Consume Apache Iggy topics as live streams/maps and produce messages as target states.

Version
v 1.0.62
Last reviewed
Jul 4, 2026

The iggy connector supports Apache Iggy as both a source and a target. Iggy organizes messages as streams → topics → partitions; CocoIndex follows that model and treats an Iggy topic partition either as a raw live stream or as an application-keyed live map.

python
from cocoindex.connectors import iggy

Install the optional dependency:

bash
pip install \
  --index-url "https://token:${COCOINDEX_PLUS_INSTALL_TOKEN}@pypi.pkg.keygen.sh/cocoindex-io/simple" \
  "cocoindex-plus[iggy]"

The connector expects an apache_iggy.IggyClient that you create, connect, and provide through your app context. Streams and topics are user-managed: CocoIndex does not create or drop them.

As Source

As a live stream

Use topic_as_stream() when every Iggy message is an event and downstream code does not need map-style deletion semantics.

python
def topic_as_stream(
    client: IggyClient,
    consumer_group: str,
    stream: str,
    topic: str,
    *,
    partition_id: int = 0,
    batch_length: int = 100,
    allow_replay: bool = False,
    initial_high_watermark: int | None = None,
) -> TopicStream

TopicStream.payloads() adapts the stream to LiveStream[bytes] when the processing logic only needs message payload bytes.

python
events = iggy.topic_as_stream(
    client,
    consumer_group="cocoindex-worker",
    stream="orders",
    topic="events",
).payloads()

As a live keyed map

Use topic_as_map() when message payloads encode an application-level key and you want CocoIndex to treat the topic as a live map.

python
def topic_as_map(
    client: IggyClient,
    consumer_group: str,
    stream: str,
    topic: str,
    *,
    key: KeyFn,
    is_deletion: IsDeleteFn | None = None,
    partition_id: int = 0,
    batch_length: int = 100,
    allow_replay: bool = False,
    initial_high_watermark: int | None = None,
) -> LiveMapFeed[StableKey, ReceiveMessage]

Iggy Python messages do not expose Kafka-style keys or tombstones, so key is required. Return None from key to skip a message. Pass is_deletion if your payload format has application-level delete events.

python
import json


def key(message) -> str | None:
    payload = json.loads(message.payload())
    return payload.get("id")


items = iggy.topic_as_map(
    client,
    consumer_group="cocoindex-worker",
    stream="orders",
    topic="events",
    key=key,
)

Readiness and offsets

The source connector disables Iggy auto-commit and stores offsets after the downstream readiness handle completes. This mirrors Kafka-style back-pressure: an offset is stored only after CocoIndex has finished processing the message.

For single-partition topics, the connector can infer the initial high watermark from Iggy’s topic details. For multi-partition topics, pass initial_high_watermark for the consumed partition; the current Python SDK does not expose per-partition high-watermark callbacks.

As Target

The target connector sends bytes or strings to a user-managed Iggy stream/topic/partition.

python
IGGY = coco.ContextKey[IggyClient]("iggy")

target = await iggy.mount_iggy_topic_target(
    IGGY,
    stream="orders",
    topic="derived-events",
    partition=0,
)

target.declare_target_state(key="order-123", value=b'{"status":"ready"}')

While the topic is still declared, a message target state that is no longer declared gets a deletion message. Deletes need an application-level delete payload because Iggy does not have Kafka-style tombstones:

python
target = await iggy.mount_iggy_topic_target(
    IGGY,
    stream="orders",
    topic="derived-events",
    deletion_value_fn=lambda key: f'{{"id":{key!r},"deleted":true}}',
)

Abandoning a topic

Removing the topic declaration itself — or calling App.drop() — abandons the topic: CocoIndex stops producing to it and sends no deletion messages for the messages it produced earlier. Changing the stream, topic, or partition in code abandons the old one the same way. Avoid reusing an abandoned topic in a later declaration: its stale keys remain and never get a deletion message, and keys you declare again are republished as new even when their values are unchanged.

To send deletion messages for everything before abandoning a topic, drain it first: keep the topic declared but declare no messages in it, run an update, then remove the declaration.

Target APIs

python
def declare_iggy_topic_target(
    client: ContextKey[IggyClient],
    stream: str,
    topic: str,
    *,
    partition: int = 0,
    deletion_value_fn: DeletionValueFn | None = None,
) -> IggyTopicTarget[PendingS]
python
async def mount_iggy_topic_target(
    client: ContextKey[IggyClient],
    stream: str,
    topic: str,
    *,
    partition: int = 0,
    deletion_value_fn: DeletionValueFn | None = None,
) -> IggyTopicTarget[ResolvedS]
PreviousGoogle Drive
CocoIndex Docs