Cdc Streaming Ingestion

CDC / Streaming Ingestion

When to Go Real-Time

Scheduled ingestion (see ingestion-orchestration) suffices when freshness tolerance is minutes-to-hours. Switch to streaming when:

• Users expect < 10s from source write to retrievable.
• Source is high-throughput (> 100 writes/sec).
• Deletes must propagate quickly (compliance).
• Downstream consumers beyond RAG also need the stream.

Architecture

[Postgres]  --WAL-->  [Debezium]  -->  [Kafka topic: docs.changes]
                                               |
                                 +-------------+-------------+
                                 |                           |
                           [embed worker]              [other consumers]
                                 |
                           [Vector DB upsert]
                                 |
                         [Kafka topic: docs.indexed]   (offset log for observability)

Separation of concerns:

• CDC produces change events.
• Kafka is the durable, replayable event log.
• Stream processor embeds and upserts.
• Vector DB is the sink.

CDC Source: Postgres + Debezium

Postgres logical replication exposes WAL changes. Debezium Connect consumes it and publishes to Kafka.

Postgres setup

ALTER SYSTEM SET wal_level = 'logical';
SELECT pg_reload_conf();

CREATE PUBLICATION rag_pub FOR TABLE documents, articles;

CREATE USER debezium WITH REPLICATION LOGIN PASSWORD '***';
GRANT SELECT ON documents, articles TO debezium;

SELECT pg_create_logical_replication_slot('debezium_rag', 'pgoutput');

Debezium connector config

{
  "name": "postgres-rag",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${secret}",
    "database.dbname": "app",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_rag",
    "publication.name": "rag_pub",
    "topic.prefix": "rag",
    "table.include.list": "public.documents,public.articles",
    "tombstones.on.delete": "true",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "heartbeat.interval.ms": "30000",
    "heartbeat.action.query": "INSERT INTO debezium_heartbeat (ts) VALUES (now()) ON CONFLICT DO NOTHING"
  }
}

Key points:

• pgoutput plugin comes with Postgres 10+; no separate install.
• tombstones.on.delete=true sends a null-value record after a delete for log-compacted downstream topics.
• Heartbeats keep the replication slot active on low-traffic tables.
• Avro + schema registry for schema evolution discipline.

Event shape (Debezium op codes: c create, u update, d delete, r snapshot-read):

{
  "before": { "id": 42, "title": "Old title", ... },
  "after":  { "id": 42, "title": "New title", ... },
  "source": { "lsn": 987654321, "txId": 445, "ts_ms": 1744726800000 },
  "op": "u",
  "ts_ms": 1744726800012
}

Stream Processor: Python + Kafka

For moderate throughput, a Python consumer with aiokafka is enough.

import asyncio, json
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from openai import AsyncOpenAI
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import PointStruct

oai = AsyncOpenAI()
qdr = AsyncQdrantClient(url="http://qdrant:6333")

async def process():
    consumer = AIOKafkaConsumer(
        "rag.public.documents",
        bootstrap_servers="kafka:9092",
        group_id="rag-embedder",
        enable_auto_commit=False,           # manual commit after successful upsert
        auto_offset_reset="earliest",
    )
    producer = AIOKafkaProducer(bootstrap_servers="kafka:9092",
                                 enable_idempotence=True, acks="all")
    await consumer.start(); await producer.start()

    try:
        async for msg in consumer:
            if msg.value is None:
                # Tombstone -> delete from vector DB
                doc_id = json.loads(msg.key.decode())["id"]
                await qdr.delete(collection_name="kb",
                                 points_selector={"filter": {"must": [
                                     {"key": "doc_id", "match": {"value": doc_id}}
                                 ]}})
                await consumer.commit(); continue

            evt = json.loads(msg.value.decode())
            op = evt["op"]
            doc = evt["after"] if op in ("c", "u", "r") else evt["before"]
            doc_id = doc["id"]

            if op == "d":
                await qdr.delete(collection_name="kb",
                                 points_selector={"filter": {"must": [
                                     {"key": "doc_id", "match": {"value": doc_id}}
                                 ]}})
            else:
                chunks = chunk_doc(doc)
                texts = [c["text"] for c in chunks]
                emb = await oai.embeddings.create(model="text-embedding-3-small", input=texts)
                points = [
                    PointStruct(
                        id=f"{doc_id}-{i}",
                        vector=emb.data[i].embedding,
                        payload={"doc_id": doc_id, "chunk_index": i,
                                 "text": chunks[i]["text"], "lsn": evt["source"]["lsn"]},
                    )
                    for i in range(len(chunks))
                ]
                # Delete old chunks then upsert new — handles chunk count change
                await qdr.delete(collection_name="kb",
                                 points_selector={"filter": {"must": [
                                     {"key": "doc_id", "match": {"value": doc_id}}
                                 ]}})
                await qdr.upsert(collection_name="kb", points=points)

            await producer.send("rag.indexed", key=msg.key,
                                value=json.dumps({"doc_id": doc_id, "op": op}).encode())
            await consumer.commit()
    finally:
        await consumer.stop(); await producer.stop()

asyncio.run(process())

• enable_auto_commit=False with manual commit after upsert => at-least-once delivery.
• Delete-then-upsert handles chunk count changes on update.
• Emit to rag.indexed topic for observability.

Exactly-Once Semantics

Kafka exactly-once (EOS) requires transactional writes. The vector DB is usually not transactional with Kafka, so you cannot get true EOS end-to-end. Alternative: make upserts idempotent.

# Use stable point IDs: f"{doc_id}-{chunk_index}"
# Use upsert (not insert) in the vector DB.
# Include LSN in payload; on reprocess, skip if stored LSN >= event LSN.

async def idempotent_upsert(point: PointStruct):
    existing = await qdr.retrieve(collection_name="kb", ids=[point.id], with_payload=True)
    if existing and existing[0].payload.get("lsn", 0) >= point.payload["lsn"]:
        return  # already applied a newer or equal version
    await qdr.upsert(collection_name="kb", points=[point])

Result: at-least-once delivery + idempotent sink = effectively exactly-once.

Late-Arriving Updates

CDC can reorder across partitions. Pin all events of a given document to the same partition (by primary key) so updates arrive in order per document.

Debezium does this by default when message.key is the table primary key. Verify:

kafka-console-consumer --topic rag.public.documents --property print.key=true
# Key: {"id":42} for all events about document 42

Cross-document ordering is not guaranteed; it doesn't matter for embedding.

Tombstones (Deletes)

Debezium produces two messages on delete:

1. The delete event (op=d, before populated, after null).
2. A tombstone (null value) for log compaction.

Handle both in the consumer:

• On op=d: delete from vector DB.
• On null value (tombstone): also delete (defensive; safe to re-delete).

Filter by op == "d" if your vector DB delete is costly:

if evt.get("op") == "d":
    ...  # delete

Flink Streaming (high throughput, stateful)

For > 10k events/sec or stateful joins (enrich from another topic before embedding), use Flink.

# Flink SQL (PyFlink)
from pyflink.table import EnvironmentSettings, TableEnvironment

env = TableEnvironment.create(EnvironmentSettings.in_streaming_mode())

env.execute_sql("""
CREATE TABLE documents (
    id BIGINT,
    title STRING,
    body STRING,
    updated_at TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'rag.public.documents',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'debezium-json',
    'scan.startup.mode' = 'earliest-offset'
)""")

env.execute_sql("""
CREATE TABLE enriched AS
SELECT d.id, d.title, d.body, u.tenant_id, u.role
FROM documents d
LEFT JOIN users FOR SYSTEM_TIME AS OF d.updated_at AS u ON d.author_id = u.id
""")
# Sink to another topic consumed by the embed worker.

debezium-json format understands the CDC envelope natively. Temporal joins enrich events with slowly changing dimensions.

Schema Evolution

1. Use Avro with a schema registry (Confluent, Redpanda, Apicurio).
2. Enforce backward-compatible changes: add optional fields only.
3. Track schema version in vector DB metadata.
4. On breaking change, publish to a new topic (rag.public.documents.v2) and run both consumers during migration.

# In the upsert payload
payload = {"doc_id": doc_id, "schema_version": "v2", "lsn": lsn, ...}

When the embedder version changes (different embedding dimension/model), do a blue-green re-index off the CDC stream — start a new consumer group on earliest, write to a new collection, switch alias.

Backpressure and Rate Limits

Embedding APIs have rate limits. Surge events can blow them. Options:

1. Batch events before embedding (collect for 1-2s or 64 docs).
2. Use a concurrency semaphore.
3. Configure consumer max.poll.records to cap per-poll batch.

import asyncio
semaphore = asyncio.Semaphore(8)

async def embed_batch(batch: list[dict]):
    async with semaphore:
        texts = [b["text"] for b in batch]
        return await oai.embeddings.create(model="text-embedding-3-small", input=texts)

Monitoring

• Consumer lag: kafka.consumer.lag per topic/partition. Alert > 1000.
• Replication slot lag: pg_replication_slots.confirmed_flush_lsn vs current WAL.
• Embedding throughput: events/sec, rolling P95.
• Vector DB upsert errors: by error class.
• End-to-end latency: now() - evt.ts_ms percentile dashboard.

-- Postgres: watch replication slot lag
SELECT slot_name, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS bytes_behind
FROM pg_replication_slots WHERE slot_name = 'debezium_rag';

An unconsumed slot holds WAL on disk. Alert above a few GB.

Anti-Patterns

Production Checklist

• Postgres wal_level=logical, publication created, replication user configured
• Debezium connector deployed with heartbeats enabled
• Schema registry (Avro) in use
• Tombstones on delete enabled
• Consumer group with manual offset commit
• Idempotent upsert with stable point IDs + LSN gate
• Batching + concurrency semaphore on embedding API
• Dead-letter topic for poison messages
• Replication slot lag alert
• Consumer group lag alert
• End-to-end latency SLO (e.g., P95 < 10s)
• Content-hash gate to skip no-op updates
• Blue-green re-index procedure for embedding model changes
• Runbook for replication slot recovery (drop + recreate from snapshot)
• Observability topic (rag.indexed) consumed by dashboards