cdc streaming ingestion

I have configured the CDC pipeline for real-time document ingestion. This setup includes the Postgres publication, the Debezium connector configuration, and a Python consumer that handles idempotent upserts to Qdrant.

Real-Time RAG Ingestion Pipeline

1. Postgres Publication and Slot

-- Enable logical replication and create publication
ALTER SYSTEM SET wal_level = 'logical';
SELECT pg_reload_conf();
CREATE PUBLICATION rag_pub FOR TABLE articles;
SELECT pg_create_logical_replication_slot('debezium_rag', 'pgoutput');

2. Debezium Connector Configuration

{
  "name": "postgres-articles-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "db-host",
    "database.port": "5432",
    "database.dbname": "app_db",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_rag",
    "publication.name": "rag_pub",
    "tombstones.on.delete": "true",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

3. Idempotent Python Consumer

async def idempotent_upsert(qdrant, point, event_lsn):
    # Retrieve existing LSN to prevent out-of-order overwrites
    existing = await qdrant.retrieve(collection_name="kb", ids=[point.id])
    if existing and existing[0].payload.get("lsn", 0) >= event_lsn:
        return 
    await qdrant.upsert(collection_name="kb", points=[point])

Next steps

• Deploy the Debezium connector to your Kafka Connect cluster.
• Monitor pg_replication_slots to ensure the slot doesn't grow indefinitely.
• Initialize the Qdrant collection with a payload index on doc_id for efficient deletes.