Test Kafka to Warehouse Data Pipeline (2026)

TL;DR

Start real Kafka and PostgreSQL instances, publish controlled events, run the actual sink logic, and assert both raw lineage and warehouse facts. Add replay, ordering, poison-event, and reconciliation checks so the suite proves data correctness rather than transport alone.

Key Takeaways

• Assert the warehouse row and source metadata, not merely a successful Kafka produce call.
• Commit a Kafka offset only after the warehouse transaction accepts or quarantines the record.
• Use an immutable event ID and a warehouse uniqueness constraint to make replays harmless.
• Test late events, malformed payloads, duplicate delivery, and poison records as separate risks.
• Reconcile raw events against current-state facts with SQL instead of trusting consumer logs.
• Run the same broker-backed checks in CI with pinned container and client versions.

To test kafka to warehouse data pipeline behavior, publish a uniquely identifiable event to a real broker, process it with production-shaped sink logic, and query the warehouse for its business values and lineage. A producer acknowledgement is only evidence that Kafka stored bytes. It says nothing about deserialization, transformation, database commit, offset handling, or analytical correctness.

This tutorial follows one controlled order through Apache Kafka, Python sink code, raw lineage, current warehouse facts, and quarantine data. You will run the same path under normal delivery, replay, late arrival, and invalid input.

What You Will Build

You will create an executable local project that:

• Runs a single-node Apache Kafka 4.2.0 broker and PostgreSQL 18.4 warehouse in Docker.
• Publishes OrderUpserted JSON events to orders.normalized.v1.
• Persists every accepted event in raw_order_events with Kafka lineage.
• Maintains one current row per order in fact_orders, protected from late arrivals.
• Records malformed or semantically invalid input in pipeline_failures.
• Executes broker-backed pytest checks and SQL reconciliation in CI.

Prerequisites

Use these pinned versions for a reproducible 2026 baseline:

• Python 3.14.6
• Docker Engine 29.6.2 with Docker Compose 5.3.1
• apache/kafka-native:4.2.0
• postgres:18.4-alpine
• confluent-kafka==2.15.0
• psycopg[binary]==3.3.4
• pytest==9.1.1

Confirm the host tools before creating files:

python3 --version
docker version
docker compose version

Create a clean directory and virtual environment:

mkdir kafka-warehouse-test && cd kafka-warehouse-test
python3 -m venv .venv
source .venv/bin/activate

Save requirements.txt:

confluent-kafka==2.15.0
psycopg[binary]==3.3.4
pytest==9.1.1

Install it with python -m pip install -r requirements.txt. Verification: Run python -c "import confluent_kafka, psycopg, pytest; print(confluent_kafka.version())". The command must exit with code 0 and print a Confluent Kafka client version tuple.

Step 1: Start Kafka and the Warehouse

Save this as compose.yaml:

services:
  kafka:
    image: apache/kafka-native:4.2.0
    ports:
      - '9092:9092'
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    healthcheck:
      test: ['CMD-SHELL', '/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list >/dev/null 2>&1']
      interval: 5s
      timeout: 5s
      retries: 20
      start_period: 10s
  warehouse:
    image: postgres:18.4-alpine
    ports:
      - '5432:5432'
    environment:
      POSTGRES_USER: qa
      POSTGRES_PASSWORD: qa
      POSTGRES_DB: analytics
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U qa -d analytics']
      interval: 2s
      timeout: 3s
      retries: 20

Start both services, then create a one-partition topic. One partition makes ordering tests deterministic; production topology can use more partitions while preserving order per key.

docker compose up -d --wait
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create --if-not-exists \
  --topic orders.normalized.v1 \
  --partitions 1 --replication-factor 1

Verification: Run docker compose ps and confirm both containers are running and PostgreSQL is healthy. Then run docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders.normalized.v1; expect PartitionCount: 1 and ReplicationFactor: 1.

Step 2: Create Warehouse Tables With Testable Invariants

Create sql/schema.sql:

CREATE TABLE IF NOT EXISTS raw_order_events (
  event_id text PRIMARY KEY,
  topic text NOT NULL,
  partition_id integer NOT NULL,
  event_offset bigint NOT NULL,
  order_id text NOT NULL,
  occurred_at timestamptz NOT NULL,
  payload jsonb NOT NULL,
  ingested_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (topic, partition_id, event_offset)
);

CREATE TABLE IF NOT EXISTS fact_orders (
  order_id text PRIMARY KEY,
  customer_id text NOT NULL,
  status text NOT NULL CHECK (status IN ('CREATED', 'PAID', 'CANCELLED')),
  amount numeric(12,2) NOT NULL CHECK (amount >= 0),
  currency char(3) NOT NULL,
  occurred_at timestamptz NOT NULL,
  source_event_id text NOT NULL UNIQUE REFERENCES raw_order_events(event_id),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS pipeline_failures (
  topic text NOT NULL,
  partition_id integer NOT NULL,
  event_offset bigint NOT NULL,
  error_reason text NOT NULL,
  payload_text text,
  failed_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (topic, partition_id, event_offset)
);

Apply the schema:

mkdir -p sql
docker compose exec -T warehouse psql -U qa -d analytics < sql/schema.sql

The event_id primary key makes a Kafka replay idempotent at the raw boundary. The topic coordinate constraint detects an impossible condition where the same Kafka position is associated with a different event ID. The fact constraints reject negative money and unsupported states even if application validation weakens.

Verification: Run docker compose exec warehouse psql -U qa -d analytics -c '\dt'. Expect raw_order_events, fact_orders, and pipeline_failures. Run \d fact_orders in psql if you need to inspect the check and uniqueness constraints.

Step 3: Publish a Controlled Kafka Event

Create fixtures/order-created.json:

{
  "eventId": "evt-1001",
  "orderId": "ord-501",
  "customerId": "cus-42",
  "status": "CREATED",
  "amount": "79.95",
  "currency": "USD",
  "occurredAt": "2026-08-06T08:30:00Z"
}

Create produce.py. The function is imported by later tests, so keep its name and signature unchanged.

import json
import os
import sys
from pathlib import Path
from confluent_kafka import Producer

BOOTSTRAP = os.getenv('KAFKA_BOOTSTRAP', 'localhost:9092')
TOPIC = 'orders.normalized.v1'

def publish_event(event: dict) -> None:
    producer = Producer({
        'bootstrap.servers': BOOTSTRAP,
        'enable.idempotence': True,
        'acks': 'all',
    })
    delivery_errors = []
    def delivered(error, _message):
        if error is not None:
            delivery_errors.append(str(error))
    producer.produce(
        TOPIC,
        key=event['orderId'].encode(),
        value=json.dumps(event).encode(),
        headers=[('event-type', b'OrderUpserted')],
        on_delivery=delivered,
    )
    remaining = producer.flush(10)
    if remaining != 0 or delivery_errors:
        raise RuntimeError(
            f'publish failed: remaining={remaining}, errors={delivery_errors}'
        )

if __name__ == '__main__':
    publish_event(json.loads(Path(sys.argv[1]).read_text()))

The event uses strings for money and ISO-8601 UTC time. That avoids binary floating-point surprises and makes the expected warehouse value explicit. The record key equals orderId, preserving per-order order when the topic has several partitions. The event ID is a delivery identity, not a mutable business key.

Verification: Run python produce.py fixtures/order-created.json. Expect exit code 0. Then run docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders.normalized.v1 --from-beginning --max-messages 1; the printed JSON must contain evt-1001 and 79.95.

Step 4: Implement the Warehouse Sink Boundary

Create pipeline.py. This is the same callable boundary used by the daemon and tests.

import json
import os
from datetime import datetime
from decimal import Decimal, InvalidOperation
import psycopg
from psycopg.types.json import Jsonb
from confluent_kafka import Consumer, KafkaError

DSN = os.getenv('WAREHOUSE_DSN', 'postgresql://qa:qa@localhost:5432/analytics')
BOOTSTRAP = os.getenv('KAFKA_BOOTSTRAP', 'localhost:9092')
TOPIC = 'orders.normalized.v1'
ALLOWED_STATUS = {'CREATED', 'PAID', 'CANCELLED'}

class PipelineError(ValueError):
    pass

def normalize(record) -> dict:
    headers = {k: v.decode() for k, v in (record.headers() or [])}
    if headers.get('event-type') != 'OrderUpserted':
        raise PipelineError('event-type must be OrderUpserted')
    payload = json.loads(record.value().decode())
    required = {
        'eventId', 'orderId', 'customerId', 'status',
        'amount', 'currency', 'occurredAt',
    }
    missing = sorted(required - payload.keys())
    if missing:
        raise PipelineError(f'missing fields: {missing}')
    if record.key().decode() != payload['orderId']:
        raise PipelineError('record key must equal orderId')
    if payload['status'] not in ALLOWED_STATUS:
        raise PipelineError('unsupported status')
    amount = Decimal(str(payload['amount'])).quantize(Decimal('0.01'))
    if amount < 0:
        raise PipelineError('amount must be nonnegative')
    if len(payload['currency']) != 3 or not payload['currency'].isalpha():
        raise PipelineError('currency must contain three letters')
    occurred_at = datetime.fromisoformat(
        payload['occurredAt'].replace('Z', '+00:00')
    )
    if occurred_at.tzinfo is None:
        raise PipelineError('occurredAt must include a timezone')
    return {**payload, 'amount': amount, 'occurred_at': occurred_at}

class WarehouseSink:
    def __init__(self, connection):
        self.connection = connection

    def process_record(self, record) -> str:
        try:
            event = normalize(record)
            with self.connection.transaction():
                inserted = self.connection.execute(
                    '''
                    INSERT INTO raw_order_events
                      (event_id, topic, partition_id, event_offset, order_id, occurred_at, payload)
                    VALUES (%s, %s, %s, %s, %s, %s, %s)
                    ON CONFLICT (event_id) DO NOTHING
                    RETURNING event_id
                    ''',
                    (event['eventId'], record.topic(), record.partition(),
                     record.offset(), event['orderId'], event['occurred_at'],
                     Jsonb(json.loads(record.value()))),
                ).fetchone()
                if inserted:
                    self.connection.execute(
                        '''
                        INSERT INTO fact_orders
                          (order_id, customer_id, status, amount, currency,
                           occurred_at, source_event_id)
                        VALUES (%s, %s, %s, %s, %s, %s, %s)
                        ON CONFLICT (order_id) DO UPDATE SET
                          customer_id = EXCLUDED.customer_id,
                          status = EXCLUDED.status,
                          amount = EXCLUDED.amount,
                          currency = EXCLUDED.currency,
                          occurred_at = EXCLUDED.occurred_at,
                          source_event_id = EXCLUDED.source_event_id,
                          updated_at = now()
                        WHERE EXCLUDED.occurred_at >= fact_orders.occurred_at
                        ''',
                        (event['orderId'], event['customerId'], event['status'],
                         event['amount'], event['currency'].upper(),
                         event['occurred_at'], event['eventId']),
                    )
            return 'accepted' if inserted else 'duplicate'
        except (PipelineError, json.JSONDecodeError, InvalidOperation,
                KeyError, TypeError, UnicodeDecodeError, ValueError) as error:
            with self.connection.transaction():
                self.connection.execute(
                    '''
                    INSERT INTO pipeline_failures
                      (topic, partition_id, event_offset, error_reason, payload_text)
                    VALUES (%s, %s, %s, %s, %s)
                    ON CONFLICT DO NOTHING
                    ''',
                    (record.topic(), record.partition(), record.offset(),
                     str(error), record.value().decode(errors='replace')),
                )
            return 'rejected'

def run() -> None:
    consumer = Consumer({
        'bootstrap.servers': BOOTSTRAP,
        'group.id': 'warehouse-orders-v1',
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,
    })
    consumer.subscribe([TOPIC])
    with psycopg.connect(DSN, autocommit=True) as connection:
        sink = WarehouseSink(connection)
        try:
            while True:
                record = consumer.poll(1.0)
                if record is None:
                    continue
                if record.error():
                    if record.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    raise RuntimeError(record.error())
                print(sink.process_record(record), record.offset())
                consumer.commit(message=record, asynchronous=False)
        finally:
            consumer.close()

if __name__ == '__main__':
    run()

The database transaction ends before the synchronous offset commit. A crash after the SQL commit but before the Kafka commit causes replay, and the event_id constraint turns that replay into duplicate. A database connectivity or constraint error is deliberately not caught as a poison event, so the offset stays uncommitted and infrastructure failure remains retryable.

The fact upsert has an event-time guard. A delayed CREATED event cannot overwrite a later PAID event. If your source supplies a trustworthy sequence number, compare that instead of timestamps, since clocks can be wrong.

Verification: Run python pipeline.py in one terminal, then publish the fixture in another. Query with docker compose exec warehouse psql -U qa -d analytics -c "SELECT order_id,status,amount,currency,source_event_id FROM fact_orders". Expect ord-501 | CREATED | 79.95 | USD | evt-1001. Stop the consumer with Ctrl+C after the row appears.

Step 5: Test Kafka to Warehouse Data Pipeline End to End

Create tests/test_pipeline.py. The consumer joins before publication and uses latest, which prevents old development records from contaminating a test.

import json
import time
import uuid
from pathlib import Path
import psycopg
from confluent_kafka import Consumer
from pipeline import DSN, TOPIC, WarehouseSink
from produce import publish_event

def started_consumer() -> Consumer:
    consumer = Consumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': f'test-{uuid.uuid4()}',
        'auto.offset.reset': 'latest',
        'enable.auto.commit': False,
    })
    consumer.subscribe([TOPIC])
    deadline = time.monotonic() + 10
    while not consumer.assignment() and time.monotonic() < deadline:
        consumer.poll(0.2)
    assert consumer.assignment(), 'consumer never received a partition'
    return consumer

def next_record(consumer: Consumer):
    deadline = time.monotonic() + 10
    while time.monotonic() < deadline:
        record = consumer.poll(0.2)
        if record is not None and not record.error():
            return record
    raise AssertionError('no Kafka record arrived within 10 seconds')

def reset_tables(connection) -> None:
    connection.execute(
        'TRUNCATE pipeline_failures, fact_orders, raw_order_events'
    )

def test_valid_event_reaches_warehouse():
    event = json.loads(Path('fixtures/order-created.json').read_text())
    consumer = started_consumer()
    try:
        publish_event(event)
        record = next_record(consumer)
        with psycopg.connect(DSN, autocommit=True) as connection:
            reset_tables(connection)
            outcome = WarehouseSink(connection).process_record(record)
            row = connection.execute(
                '''
                SELECT f.order_id, f.status, f.amount::text, f.currency,
                       r.topic, r.partition_id, r.event_offset
                FROM fact_orders f
                JOIN raw_order_events r ON r.event_id = f.source_event_id
                WHERE f.order_id = %s
                ''',
                (event['orderId'],),
            ).fetchone()
        assert outcome == 'accepted'
        assert row[:4] == ('ord-501', 'CREATED', '79.95', 'USD')
        assert row[4:] == (TOPIC, record.partition(), record.offset())
    finally:
        consumer.close()

This test asserts business columns and Kafka lineage in one query. It does not call the always-running run() loop because process lifetime is not the behavior under test. It invokes the same WarehouseSink.process_record method the loop uses, with a genuine Message returned by Kafka.

Verification: Run pytest -q tests/test_pipeline.py::test_valid_event_reaches_warehouse. Expect 1 passed. Change the fixture amount to 79.955 and decide whether rounding is permitted; the current normalization deliberately stores 79.96, so encode the product rule in the assertion rather than accepting accidental behavior.

Step 6: Reconcile Kafka Lineage With Warehouse Facts

A fixture proves one path. Reconciliation SQL examines the accumulated dataset and catches drift caused by a code path your test did not predict. Save sql/reconcile.sql:

-- Accepted raw events whose current order has no fact row.
SELECT r.event_id, r.order_id, r.occurred_at
FROM raw_order_events r
LEFT JOIN fact_orders f ON f.order_id = r.order_id
WHERE f.order_id IS NULL;

-- Fact rows that do not point to the newest accepted event for the order.
WITH ranked AS (
  SELECT event_id, order_id, occurred_at,
         row_number() OVER (
           PARTITION BY order_id
           ORDER BY occurred_at DESC, event_offset DESC
         ) AS position
  FROM raw_order_events
)
SELECT f.order_id, f.source_event_id, r.event_id AS expected_event_id
FROM fact_orders f
JOIN ranked r ON r.order_id = f.order_id AND r.position = 1
WHERE f.source_event_id <> r.event_id;

-- Accepted events with suspicious ingestion delay, using an illustrative limit.
SELECT event_id, occurred_at, ingested_at,
       ingested_at - occurred_at AS ingestion_delay
FROM raw_order_events
WHERE ingested_at - occurred_at > interval '15 minutes';

The first two queries should return zero rows. The third is contextual: backfills legitimately arrive late, so do not turn a sample 15-minute threshold into a universal service objective. Segment live traffic from declared backfills or compare against a pipeline-specific freshness agreement.

Verification: Run docker compose exec -T warehouse psql -U qa -d analytics < sql/reconcile.sql. After only the tutorial fixture, the missing and mismatched queries must print zero rows. The delay query may show the fixed historical fixture, which demonstrates why freshness thresholds need test data anchored to the current clock.

Step 7: Extend the Test Kafka to Warehouse Data Pipeline Suite

Add replay, late-arrival, and poison-event tests below the happy-path test. These cases exercise independent correctness properties.

def deliver(event: dict, consumer: Consumer, sink: WarehouseSink):
    publish_event(event)
    return sink.process_record(next_record(consumer))

def test_replay_is_idempotent():
    event = json.loads(Path('fixtures/order-created.json').read_text())
    consumer = started_consumer()
    try:
        publish_event(event)
        record = next_record(consumer)
        with psycopg.connect(DSN, autocommit=True) as connection:
            reset_tables(connection)
            sink = WarehouseSink(connection)
            assert sink.process_record(record) == 'accepted'
            assert sink.process_record(record) == 'duplicate'
            counts = connection.execute(
                'SELECT (SELECT count(*) FROM raw_order_events), '
                '(SELECT count(*) FROM fact_orders)'
            ).fetchone()
        assert counts == (1, 1)
    finally:
        consumer.close()

def test_late_event_cannot_replace_newer_state():
    older = json.loads(Path('fixtures/order-created.json').read_text())
    newer = {**older, 'eventId': 'evt-1002', 'status': 'PAID',
             'occurredAt': '2026-08-06T08:35:00Z'}
    older = {**older, 'eventId': 'evt-1003'}
    consumer = started_consumer()
    try:
        with psycopg.connect(DSN, autocommit=True) as connection:
            reset_tables(connection)
            sink = WarehouseSink(connection)
            assert deliver(newer, consumer, sink) == 'accepted'
            assert deliver(older, consumer, sink) == 'accepted'
            state = connection.execute(
                'SELECT status, source_event_id FROM fact_orders '
                'WHERE order_id = %s', ('ord-501',)
            ).fetchone()
        assert state == ('PAID', 'evt-1002')
    finally:
        consumer.close()

def test_invalid_currency_is_quarantined():
    base = json.loads(Path('fixtures/order-created.json').read_text())
    invalid = {**base, 'eventId': 'evt-bad', 'currency': 'US'}
    consumer = started_consumer()
    try:
        with psycopg.connect(DSN, autocommit=True) as connection:
            reset_tables(connection)
            outcome = deliver(invalid, consumer, WarehouseSink(connection))
            reason = connection.execute(
                'SELECT error_reason FROM pipeline_failures'
            ).fetchone()[0]
            fact_count = connection.execute(
                'SELECT count(*) FROM fact_orders'
            ).fetchone()[0]
        assert outcome == 'rejected'
        assert 'three letters' in reason
        assert fact_count == 0
    finally:
        consumer.close()

Replay calls the sink twice with the exact Kafka coordinate, simulating a crash window without manufacturing a fake message object. The late-event case uses two distinct event IDs because both records belong in the raw audit table even though only the newer one owns the current fact. The poison case proves observable rejection and absence of partial business data.

Add a database outage test separately. Stop PostgreSQL after polling a record, assert process_record raises, restart the database, and process the uncommitted record again. Do not classify an unavailable warehouse as bad input. That distinction prevents permanent data loss during a temporary infrastructure incident.

Verification: Run pytest -q. Expect four passing tests. Then remove the timestamp WHERE clause from the fact upsert; the late-event test must fail with CREATED and evt-1003, proving the assertion can detect regression.

Step 8: Run the Pipeline Test in CI

Save .github/workflows/kafka-warehouse-test.yml:

name: kafka-warehouse-pipeline
on:
  pull_request:
  push:
    branches: [main]
jobs:
  integration:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.14.6'
          cache: pip
      - run: python -m pip install -r requirements.txt
      - run: docker compose up -d --wait
      - name: Create topic
        run: >-
          docker compose exec -T kafka /opt/kafka/bin/kafka-topics.sh
          --bootstrap-server localhost:9092 --create --if-not-exists
          --topic orders.normalized.v1 --partitions 1 --replication-factor 1
      - name: Apply warehouse schema
        run: docker compose exec -T warehouse psql -U qa -d analytics < sql/schema.sql
      - run: pytest -q
      - name: Print service logs on failure
        if: failure()
        run: docker compose logs --no-color

CI should build the same topology developers run. Avoid a permanently shared topic and database because retained messages, committed groups, and parallel builds create order-dependent results. At larger scale, create unique topic names and database schemas per build, then delete them through a bounded cleanup job.

Keep contract checks faster than this integration layer. A producer team can validate record shape using consumer-driven Kafka contract testing with Pact, while the pipeline owner retains the broker and warehouse suite. Those layers answer different questions and should fail with different diagnostics.

Verification: Push a branch and inspect the workflow. All four tests should pass. As a sensitivity check, temporarily change the compose PostgreSQL password without changing the DSN; pytest must fail and the final step must display connection diagnostics. Restore the password before merging.

Troubleshooting

Problem: Kafka reports that no broker is available -> Confirm port 9092 is free, inspect docker compose logs kafka, and verify KAFKA_ADVERTISED_LISTENERS is reachable from the Python process. Host-run tests need localhost; container-run tests need a Compose service hostname and a separate internal listener.

Problem: The consumer waits even though the fixture was published -> Start the test consumer and wait for partition assignment before producing. A latest consumer that joins after publication will correctly skip that record. For a deliberate backlog test, use a unique group with earliest.

Problem: PostgreSQL says the transaction is aborted -> Use autocommit=True with explicit connection.transaction() blocks as shown. After an uncaught SQL error in a manual transaction, roll it back before issuing another statement. Do not catch a constraint error and continue inside the failed transaction.

Problem: Replayed records create two business effects -> Enforce uniqueness on immutable event_id in the database and only run the fact mutation when the raw insert returns a row. An in-memory set cannot protect multiple consumers or survive a restart.

Problem: An older event overwrites a newer warehouse row -> Compare a source sequence, logical version, or trusted event timestamp in the upsert. Record arrival order is not business order when producers retry, partitions change, or historical data is backfilled.

Problem: A malformed record blocks the partition forever -> Persist its topic coordinate, payload, and classification in a quarantine table or dead-letter topic, then commit according to an explicit error policy. Test recovery and replay permissions so quarantine does not become invisible data loss.

Where To Go Next

Strengthen the producer boundary with the Kafka consumer contract test walkthrough and the Pact approach to Kafka contracts. These catch field, header, and semantic incompatibilities before a broker-backed warehouse test runs.

Expand warehouse assurance with SQL validation for ETL jobs and data integrity checks using SQL. For suites that create many orders, apply the cleanup and isolation techniques in SQL test data setup and teardown.

Finally, add production observability: consumer lag by group and partition, accepted and rejected event counts, database commit latency, quarantine age, and reconciliation mismatches. Alert on a breached service objective, not on the existence of any lag, because maintenance and controlled backfills can produce expected temporary backlog.

Interview Questions and Answers

Q: What proves that a Kafka-to-warehouse pipeline works?

A successful test correlates a controlled Kafka event with the durable warehouse row and verifies transformed business values plus source coordinates. Producer acknowledgement and zero lag are supporting signals, not proof of analytical correctness.

Q: Why commit the Kafka offset after the database transaction?

Committing first can lose data if the process crashes before the warehouse write. Writing first can cause replay if the crash happens before the offset commit, so the database operation must be idempotent.

Q: How do you test duplicate delivery?

Process the same immutable event ID and Kafka message twice. Assert one raw event, one current fact, and no second business side effect.

Q: How should a pipeline handle late events?

Retain the accepted event in the raw audit layer, but update current state only when the event version or trusted timestamp is not older than the stored value. Test a newer-then-older sequence explicitly.

Q: What is the role of reconciliation SQL?

It compares independently derived expectations across ingestion and warehouse layers. Reconciliation can reveal missing, duplicated, mismatched, or stale data beyond the few examples encoded in integration tests.

Q: Should database failures go to a dead-letter queue?

Not automatically. Invalid records are candidates for quarantine, but infrastructure failures are usually retryable and should leave offsets uncommitted. Mixing the two categories can discard valid data during an outage.

Best Practices

• Give every event an immutable ID and enforce it with a warehouse constraint.
• Preserve topic, partition, offset, event time, and ingestion time for auditability.
• Validate money as decimal data and require timezone-aware timestamps.
• Separate invalid input from retryable Kafka or database infrastructure faults.
• Assert warehouse state with SQL and include a negative assertion for partial writes.
• Use bounded polling based on observable readiness, never a fixed long sleep.
• Test replay and late arrival even if the happy path is already covered.
• Pin images and clients, then run the entire suite during upgrades.
• Isolate topic groups and database state across parallel CI jobs.
• Keep reconciliation queries versioned beside the pipeline code.

Conclusion

A reliable way to test kafka to warehouse data pipeline behavior is to cross the real boundaries: broker record, transformation code, database transaction, offset commit, and SQL result. The happy path establishes connectivity, while replay, late-event, and quarantine cases establish whether the pipeline remains correct when delivery is imperfect.

Build the raw lineage layer first, make the warehouse write idempotent, and execute the pytest and reconciliation checks in every pull request. That evidence lets QA distinguish transport success from trustworthy data and gives operators precise coordinates when a production record needs investigation.

Interview Questions and Answers

Frequently Asked Questions

Related Guides

• How to Add CI to a test framework (2026)
• How to Add logging to a test framework (2026)
• How to Add reporting to a test framework (2026)
• How to Choose Test Data Management Tool (2026)
• How to Design a test data strategy (2026)
• How to Use Gatling to Load Test a GraphQL API (2026)