Building Data Pipelines with Apache Kafka

Why Kafka Became the Backbone of Data Pipelines

Before Kafka, data pipelines were a tangled web of point-to-point connections. The orders database pushed data to the analytics warehouse via a nightly batch job. The user service synced to the search index every hour. The payment system exported CSVs to the fraud detection team. Each connection was a custom script maintained by whoever wrote it, running on a cron schedule that nobody remembered setting up.

This architecture has a name: spaghetti integration. Adding a new data consumer means writing another custom extraction script. Changing the source schema breaks every downstream consumer. There is no central place to see what data flows where or whether any pipeline is broken.

Kafka solves this by placing a central, durable event bus between all data sources and consumers. Sources publish changes to Kafka topics. Consumers read from those topics at their own pace. The source does not know or care who consumes its data. Adding a new consumer is just another reader on the same topic. Kafka retains data for days or weeks, so a new consumer can start from the beginning and catch up to the present.

Think of it like a public bulletin board in a town square. Anyone can post a notice (produce), and anyone can read it (consume). The poster does not need to personally deliver the notice to each reader. New readers can go back and read all previous notices. The board does not care how many people read each notice.

Change Data Capture: Streaming Database Changes in Real-Time

The most powerful pattern in modern data pipelines is Change Data Capture (CDC). Instead of periodically querying a database for new or changed records (which is expensive, slow, and misses deletes), CDC reads the database’s internal change log and streams every insert, update, and delete as an event.

Every relational database maintains an internal log of changes for crash recovery and replication. PostgreSQL has the write-ahead log (WAL). MySQL has the binary log (binlog). CDC tools tap into these existing logs, which means they capture every change with zero impact on the database’s performance and zero risk of missing changes.

Debezium: The Gold Standard for CDC

Debezium is an open-source CDC platform built on Kafka Connect. It supports PostgreSQL, MySQL, MongoDB, SQL Server, Oracle, and more. For each table you configure, Debezium:

1. Takes an initial snapshot of the existing data.
2. Reads the database’s change log for all subsequent changes.
3. Publishes each change as a structured event to a Kafka topic (one topic per table by default).

The change events include the before-state (for updates and deletes), the after-state (for inserts and updates), and metadata about the source (database, schema, table, transaction ID, timestamp).

# Deploy a Debezium PostgreSQL connector via Kafka Connect REST API
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-cdc-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres-primary",
      "database.port": "5432",
      "database.user": "debezium_user",
      "database.password": "secure_password",
      "database.dbname": "ecommerce",
      "table.include.list": "public.orders,public.customers,public.products",
      "topic.prefix": "cdc.ecommerce",
      "plugin.name": "pgoutput",
      "slot.name": "debezium_slot",
      "publication.name": "debezium_pub",
      "snapshot.mode": "initial",
      "tombstones.on.delete": true,
      "key.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter.schema.registry.url": "http://schema-registry:8081",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "value.converter.schema.registry.url": "http://schema-registry:8081"
    }
  }'

This single configuration replaces hundreds of lines of custom ETL code. When a row is inserted into the orders table, Debezium publishes an event like this to the cdc.ecommerce.public.orders topic:

{
  "before": null,
  "after": {
    "order_id": 42,
    "customer_id": 789,
    "total": 109.97,
    "status": "created",
    "created_at": "2026-07-12T10:30:00Z"
  },
  "source": {
    "version": "2.5.0",
    "connector": "postgresql",
    "name": "cdc.ecommerce",
    "ts_ms": 1720780200000,
    "db": "ecommerce",
    "schema": "public",
    "table": "orders",
    "txId": 12345,
    "lsn": 987654321
  },
  "op": "c",
  "ts_ms": 1720780200100
}

The op field indicates the operation type: c for create (insert), u for update, d for delete, and r for read (during the initial snapshot). Downstream consumers use this to apply the correct operation to their data stores.

Kafka Connect: The Pipeline Framework

Kafka Connect is the framework that Debezium and dozens of other connectors run on. It handles the operational complexity of running connectors: task distribution, offset tracking, fault tolerance, and scaling. You focus on configuration; Kafka Connect handles execution.

Connectors come in two flavors. Source connectors read data from external systems and write it to Kafka topics (Debezium is a source connector). Sink connectors read data from Kafka topics and write it to external systems (databases, data warehouses, search indexes, object stores).

Sink Connectors: From Kafka to Everywhere

Once data is in Kafka, sink connectors deliver it to any destination. Here are the most common production patterns.

Kafka to Amazon S3

Archiving raw events to S3 creates a data lake that can be queried by Athena, Spark, or Presto. The S3 sink connector writes partitioned Parquet or JSON files organized by date.

{
  "name": "s3-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "cdc.ecommerce.public.orders",
    "s3.bucket.name": "data-lake-raw",
    "s3.region": "us-east-1",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "partition.duration.ms": "3600000",
    "locale": "en-US",
    "timezone": "UTC",
    "flush.size": "10000",
    "rotate.interval.ms": "600000"
  }
}

Kafka to Elasticsearch

Streaming data to Elasticsearch enables real-time full-text search and analytics dashboards. The Elasticsearch sink connector maps Kafka message keys to document IDs, making updates and deletes idempotent.

{
  "name": "elasticsearch-sink-orders",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "2",
    "topics": "cdc.ecommerce.public.orders",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "key.ignore": false,
    "schema.ignore": true,
    "behavior.on.null.values": "delete",
    "write.method": "upsert",
    "batch.size": 500,
    "max.buffered.records": 5000,
    "linger.ms": 1000
  }
}

The behavior.on.null.values: delete setting handles tombstone messages from Debezium. When a row is deleted from the source database, Debezium publishes a message with a null value. The Elasticsearch connector translates this into a document deletion, keeping the search index in sync.

Building a Complete Pipeline: End to End

Let us walk through a complete data pipeline that many organizations build. The goal is to stream changes from an operational database to a data warehouse for analytics, while simultaneously keeping a search index updated.

PostgreSQL (source of truth)
    |
    v
Debezium CDC (source connector)
    |
    v
Kafka Topics (cdc.ecommerce.*)
    |
    |---> S3 Sink Connector ---> Parquet files in S3
    |                               |
    |                               v
    |                           Spark/Flink batch processing
    |                               |
    |                               v
    |                           BigQuery / Snowflake (analytics)
    |
    |---> Elasticsearch Sink ---> Search index (product search)
    |
    |---> Custom consumer ---> Redis cache (hot data for API)
    |
    |---> Custom consumer ---> Fraud detection (real-time scoring)

This architecture has several powerful properties. The source database is touched only once by Debezium, regardless of how many downstream consumers exist. Adding a new destination requires zero changes to the source or to existing pipelines. If the fraud detection consumer goes down, events accumulate in Kafka and are processed when it recovers. The S3 archive provides a complete history that can be reprocessed if the warehouse schema changes.

Stream Processing: Transforming Data in Flight

Sometimes you need to transform data between source and sink. Raw CDC events might contain sensitive fields that should be masked, denormalized fields that should be flattened, or multiple event types that should be routed to different topics.

Kafka Streams or Apache Flink can sit between Kafka topics to perform this transformation:

# Stream processing example: enrich and transform CDC events
from confluent_kafka import Consumer, Producer
import json

consumer = Consumer({
    "bootstrap.servers": "kafka-1:9092",
    "group.id": "pipeline-transformer",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["cdc.ecommerce.public.orders"])

producer = Producer({"bootstrap.servers": "kafka-1:9092"})

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None or msg.error():
            continue

        cdc_event = json.loads(msg.value())
        after = cdc_event.get("after")
        if after is None:
            continue  # Skip deletes for this pipeline

        # Transform: flatten, mask PII, add derived fields
        transformed = {
            "order_id": after["order_id"],
            "customer_id": after["customer_id"],
            "total": after["total"],
            "status": after["status"],
            "order_date": after["created_at"][:10],  # Extract date
            "amount_tier": classify_amount(after["total"]),
            "source_table": cdc_event["source"]["table"],
            "cdc_timestamp": cdc_event["ts_ms"]
        }

        # Route to different topics based on content
        topic = "orders.high-value" if after["total"] > 500 else "orders.standard"
        producer.produce(topic, key=str(after["order_id"]).encode(),
                        value=json.dumps(transformed).encode())
        producer.flush()
        consumer.commit(message=msg)
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

def classify_amount(total):
    if total > 1000: return "premium"
    if total > 100: return "standard"
    return "basic"

Schema Evolution: Changing the Shape of Your Data

In a long-running pipeline, schemas change. A new column is added to the source table. A field is renamed. A field type changes from integer to string. Without a strategy for schema evolution, these changes break downstream consumers.

The Confluent Schema Registry is the standard solution. It stores versioned schemas (Avro, Protobuf, or JSON Schema) for each topic and enforces compatibility rules that prevent breaking changes.

Compatibility Modes Explained

Think of schema compatibility like the difference between renovating a house while people live in it versus demolishing and rebuilding. You need rules about what changes are safe to make while existing tenants (consumers) are still there.

• BACKWARD compatible: New schema can read data written with the previous schema. You can add optional fields (with defaults) and remove fields. This is the most common mode. New consumers can handle old data.
• FORWARD compatible: Old schema can read data written with the new schema. You can remove optional fields and add fields. Old consumers can handle new data.
• FULL compatible: Both backward and forward compatible. The safest option but the most restrictive.
• NONE: No compatibility checks. Anything goes. This is dangerous in production pipelines.

# Set compatibility mode for a subject
curl -X PUT http://schema-registry:8081/config/cdc.ecommerce.public.orders-value \
  -H "Content-Type: application/json" \
  -d '{"compatibility": "BACKWARD"}'

# Register a new schema version (will be rejected if incompatible)
curl -X POST http://schema-registry:8081/subjects/cdc.ecommerce.public.orders-value/versions \
  -H "Content-Type: application/json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"order_id\",\"type\":\"int\"},{\"name\":\"total\",\"type\":\"double\"},{\"name\":\"status\",\"type\":\"string\"},{\"name\":\"priority\",\"type\":[\"null\",\"string\"],\"default\":null}]}"
  }'

The new priority field has a default value of null, making it backward compatible. Old messages without a priority field will be read with null as the default. If you tried to add a required field without a default, the Schema Registry would reject the registration.

Error Handling: Dead Letter Queues and Retry Topics

In any data pipeline, some records will fail to process. A malformed CDC event, a network timeout writing to the sink, a schema mismatch. Without error handling, a single bad record can block the entire pipeline. The consumer retries it forever, and all subsequent records pile up.

Dead Letter Queues (DLQs)

A dead letter queue is a separate Kafka topic where failed records are sent for later investigation. After a configurable number of retries, the pipeline moves the problematic record to the DLQ and continues processing the rest. This prevents one bad apple from spoiling the barrel.

Kafka Connect has built-in DLQ support:

{
  "name": "elasticsearch-sink-with-dlq",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "cdc.ecommerce.public.orders",
    "connection.url": "http://elasticsearch:9200",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.elasticsearch-orders",
    "errors.deadletterqueue.topic.replication.factor": 3,
    "errors.deadletterqueue.context.headers.enable": true,
    "errors.retry.delay.max.ms": 60000,
    "errors.retry.timeout": 300000,
    "errors.log.enable": true,
    "errors.log.include.messages": true
  }
}

With errors.tolerance: all, the connector will not stop on errors. Failed records go to dlq.elasticsearch-orders with headers that include the error message, the connector name, and the stage where the failure occurred. An operator can inspect the DLQ, fix the root cause, and replay the failed records.

Retry Topics: Gradual Back-Off

For custom consumers, a retry topic pattern provides more control than a simple DLQ. Failed records are sent to a retry topic with increasing delays. This handles transient failures (like a temporary network issue) without losing messages or blocking the pipeline.

import json
import time
from confluent_kafka import Consumer, Producer

RETRY_TOPICS = [
    "orders.retry-1",   # Retry after 1 minute
    "orders.retry-2",   # Retry after 5 minutes
    "orders.retry-3",   # Retry after 15 minutes
]
RETRY_DELAYS = [60, 300, 900]  # seconds
DLQ_TOPIC = "orders.dlq"

producer = Producer({"bootstrap.servers": "kafka-1:9092"})

def process_with_retry(msg, retry_count=0):
    """Process a message, routing to retry topics on failure."""
    try:
        record = json.loads(msg.value())
        process_record(record)  # Your business logic
    except TransientError as e:
        # Transient failure -- retry with back-off
        if retry_count < len(RETRY_TOPICS):
            headers = [
                ("retry-count", str(retry_count + 1).encode()),
                ("original-topic", b"orders"),
                ("error", str(e).encode()),
                ("next-retry-at", str(
                    int(time.time()) + RETRY_DELAYS[retry_count]
                ).encode()),
            ]
            producer.produce(RETRY_TOPICS[retry_count],
                           key=msg.key(), value=msg.value(),
                           headers=headers)
        else:
            # Exhausted all retries -- send to DLQ
            send_to_dlq(msg, e)
    except PermanentError as e:
        # Permanent failure -- go directly to DLQ
        send_to_dlq(msg, e)

def send_to_dlq(msg, error):
    headers = [
        ("original-topic", msg.topic().encode()),
        ("error", str(error).encode()),
        ("failed-at", str(int(time.time())).encode()),
    ]
    producer.produce(DLQ_TOPIC, key=msg.key(),
                    value=msg.value(), headers=headers)
    producer.flush()

Monitoring Pipeline Health and Data Freshness

A data pipeline is only as valuable as the freshness of its data. A dashboard showing day-old data might as well be a spreadsheet. Monitoring pipeline health means tracking not just whether the pipeline is running, but how fresh the data is at each stage.

Key Pipeline Metrics

End-to-end latency: The time between a change in the source database and that change being available in the sink. For a CDC pipeline, this includes Debezium capture latency, Kafka transit time, and sink connector write latency. A healthy pipeline delivers changes in under 10 seconds. A pipeline with growing latency has a bottleneck somewhere.

Throughput at each stage: Compare the record rate at the source, in Kafka, and at the sink. If the source produces 1,000 records per second and the sink only writes 500 per second, the sink is the bottleneck and lag will grow.

DLQ volume: The number of records in dead letter queues over time. A steady trickle might indicate a known data quality issue. A sudden spike usually means something broke — a schema change, a sink outage, or a connector bug.

Connector task status: Kafka Connect exposes the status of each connector and its tasks via the REST API. A task in a “FAILED” state needs immediate attention.

# Check connector status
curl -s http://localhost:8083/connectors/postgres-cdc-connector/status | jq .

# Response shows task-level status
# {
#   "name": "postgres-cdc-connector",
#   "connector": { "state": "RUNNING", "worker_id": "connect-1:8083" },
#   "tasks": [
#     { "id": 0, "state": "RUNNING", "worker_id": "connect-1:8083" }
#   ]
# }

# Check all failed connectors in one command
curl -s http://localhost:8083/connectors | jq -r '.[]' | while read connector; do
  status=$(curl -s "http://localhost:8083/connectors/$connector/status" \
    | jq -r '.connector.state')
  if [ "$status" != "RUNNING" ]; then
    echo "ALERT: $connector is $status"
  fi
done

Data Freshness Monitoring

The most practical freshness check is to compare the timestamp of the latest record in the sink with the current time. If the gap exceeds your SLA, the pipeline is stale.

# Freshness check: compare sink data timestamp with current time
from datetime import datetime, timedelta
import psycopg2

def check_pipeline_freshness(table, max_staleness_minutes=5):
    """Alert if the latest record in the sink is too old."""
    conn = psycopg2.connect("dbname=warehouse")
    with conn.cursor() as cur:
        cur.execute(f"""
            SELECT MAX(cdc_timestamp) as latest
            FROM {table}
        """)
        latest = cur.fetchone()[0]

    if latest is None:
        return {"status": "error", "message": "No data in sink table"}

    staleness = datetime.utcnow() - latest
    if staleness > timedelta(minutes=max_staleness_minutes):
        return {
            "status": "stale",
            "staleness_minutes": staleness.total_seconds() / 60,
            "message": f"Data is {staleness.total_seconds() / 60:.1f} min old"
        }
    return {"status": "fresh", "staleness_seconds": staleness.total_seconds()}

Next Steps

Building data pipelines with Kafka is more about operational discipline than code complexity. The tools — Kafka Connect, Debezium, sink connectors — handle the heavy lifting. Your job is to configure them correctly, monitor them continuously, and have a clear error handling strategy for when things go wrong. Every pipeline will fail at some point. The question is whether you notice and recover in minutes or in hours.

• Monitoring and Operations — monitor the Kafka cluster that powers your pipelines
• KRaft and Cluster Management — manage the infrastructure your pipelines run on