coldum

Coldum is a Kafka-inspired distributed log built in Python.

Python version (pyenv)

pyenv install -s 3.12
pyenv local 3.12
python --version

Run single broker

# install uv via mise (python is managed by pyenv)
mise install

# create project venv and install deps
mise x -- uv sync --extra dev

# run broker
COLDUM_DATA_DIR=./data mise x -- uv run coldum-broker

Common commands

# tests
mise x -- uv run pytest -q

# lint
mise x -- uv run ruff check .

# type-check
mise x -- uv run mypy coldum

API

• POST /v1/topics
• GET /v1/topics/{topic}
• POST /v1/produce
• POST /v1/fetch
• POST /v1/consumer-groups/{group}/subscribe
• POST /v1/consumer-groups/{group}/heartbeat
• POST /v1/consumer-groups/{group}/commit
• GET /v1/consumer-groups/{group}/assignments
• GET /v1/consumer-groups/{group}/lag
• GET /v1/cluster/metadata
• GET /v1/metrics
• GET /v1/health

Async SDK

from coldum.client import AsyncColdumConsumer, AsyncColdumProducer


async def main() -> None:
    async with AsyncColdumProducer("http://127.0.0.1:8080") as producer:
        await producer.create_topic("events", partitions=2)
        await producer.send("events", values=[b"hello"], partition=0)

    async with AsyncColdumConsumer(
        "http://127.0.0.1:8080",
        group_id="workers",
        member_id="worker-1",
    ) as consumer:
        await consumer.subscribe(["events"], offset_reset="earliest")
        async for record in consumer.stream(wait_ms=1_000):
            print(record.topic, record.partition, record.offset, record.value)
            await consumer.commit([(record.topic, record.partition, record.offset + 1)])

stream() fetches assigned partitions concurrently and subscribe() starts a background heartbeat task so the member stays in its consumer group while the stream runs.

Notes

• Delivery guarantee is at-least-once.
• Offsets are int64 logical offsets.
• Default retention is 7 days.