Mastering the Transactional Outbox Pattern: PostgreSQL CDC, Debezium, and Kafka at Scale

# Mastering the Transactional Outbox Pattern: PostgreSQL CDC, Debezium, and Kafka at Scale

Summary: The transactional outbox pattern solves the dual-write problem in distributed architectures by atomically persisting application state changes and outbox event records within a single PostgreSQL ACID transaction. Debezium streams these events asynchronously by decoding PostgreSQL Write-Ahead Log (WAL) changes via the pgoutput logical decoding plugin, routing them to Apache Kafka with the Outbox Event Router Single Message Transformation (SMT). Downstream microservices achieve deterministic consistency through partitioned outbox schemas and idempotent consumers.

---

1. The Anatomy of the Dual-Write Problem in Distributed Systems

In an event-driven microservices architecture, a business operation rarely exists in a vacuum. When an application service mutates state—such as creating an order, debiting an account, or registering a customer—it must publish a corresponding domain event to an event broker like Apache Kafka to alert downstream systems.

The core challenge stems from executing two distinct state updates across disparate storage boundaries without a distributed coordinator:

1. Writing state to an ACID-compliant relational database (such as PostgreSQL).

2. Publishing a serialized record to a distributed commit log (such as Apache Kafka).

Because these systems reside on distinct physical networks and manage separate transaction protocols, wrapping them in a standard application-level transaction is impossible. Attempting a naive "dual write" inside application code introduces three critical distributed failure modes.

       [ Client Request: POST /orders ]
                     |
                     v
         [ Order Service Instance ]
         /                        \
        /                          \
(Write 1: SQL Tx)           (Write 2: Kafka Produce)
      v                                    v
[ PostgreSQL ]                       [ Apache Kafka ]
(State: COMMITTED)                   (Status: TIMEOUT / FAILED)
        \                                  /
         \-----> CRITICAL DATA DRIFT <----/

The Three Inevitable Failure Modes of Direct Dual Writes

1. The Database-First Race (Ghost Orders and Lost Events)

The application commits the database transaction first. Immediately afterward, it attempts to produce an event to Kafka.

If the Kafka broker is experiencing an election, the network partition drops the TCP socket, or the container process crashes due to an out-of-memory (OOM) signal before the Kafka producer acknowledgment arrives, the database commit stands permanently. However, the event was never published. Downstream inventory, billing, and fulfillment services never receive notification of the order. The business state is corrupt.

2. The Broker-First Race (Phantom Side Effects)

To prevent lost events, a developer might invert the sequence: produce the Kafka event first, then commit the database transaction.

If the database commit subsequently fails—due to a serialization anomaly (40001), a foreign key or unique constraint violation, disk space exhaustion, or a database connection pool timeout—the event has already been published to Kafka and consumed by downstream services. Downstream systems may charge a credit card or ship goods for an order that does not exist in the primary database.

3. Concurrent Ordering Inversion

Suppose two threads concurrently process status updates for aggregate root Order #1042. Thread $T_1$ transitions the order status to CANCELLED, and Thread $T_2$ transitions it to REFUNDED.

In PostgreSQL, $T_1$ commits at timestamp $t_1$, followed by $T_2$ at timestamp $t_2$. However, due to thread scheduling, CPU preemption, or network jitter, Thread $T_2$ produces its Kafka message first, and Thread $T_1$ produces second. Downstream consumers consume the messages in reverse chronological order, resetting the terminal state from REFUNDED back to CANCELLED.

Why Two-Phase Commit (2PC) and Polling Publishers Break at Scale

Two historical patterns are frequently proposed to bypass direct dual writes: Two-Phase Commit (2PC / XA) and Database Polling. Both present severe structural bottlenecks in high-throughput environments:

• Two-Phase Commit (2PC/XA): 2PC requires an external transaction manager coordinating both PostgreSQL and the message broker. Kafka does not natively support XA distributed transactions. Even where distributed coordinators exist, 2PC is a blocking protocol. Transaction duration scales with the slowest participant, and any coordinator failure holds row-level locks on primary database tables indefinitely, degrading throughput and causing cascading connection pool exhaustion.
• Polling Publishers (`SELECT ... FOR UPDATE SKIP LOCKED`): In this model, events are written to an outbox table, and a background worker periodically polls the table for unprocessed records. At low transaction volumes, polling functions adequately. However, as write throughput scales, polling workers induce severe relational database degradation: continuous B-tree index traversals, MVCC dead tuple accumulation from rapid status updates or row deletions, table and row lock contention, and cache eviction within the buffer pool. Furthermore, minimizing event dissemination latency requires shortening polling sleep intervals, which drives continuous CPU and I/O consumption even when the outbox queue is empty.

---

2. Architectural Blueprint: PostgreSQL CDC, Debezium, and Kafka

The modern solution for mission-critical enterprise systems is the Transactional Outbox Pattern powered by Change Data Capture (CDC). Instead of requiring the application to interact with Kafka directly, the application confines its operations strictly to PostgreSQL.

Inside a single PostgreSQL transaction, the application writes both the operational entity mutation (e.g., an insert into orders) and a corresponding event record into an outbox_events table. Because both operations occur within the same ACID transaction boundary, they either succeed together or rollback completely.

+-----------------------------------------------------------------------------------------+
|                                    APPLICATION LAYER                                    |
|                                                                                         |
|  BEGIN TRANSACTION;                                                                     |
|    INSERT INTO orders (id, customer_id, total_amount) VALUES (...);                     |
|    INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload)    |
|         VALUES ('9c8f...', 'Order', 'ord_123', 'OrderCreated', '{"amount": 99.00}');   |
|  COMMIT;                                                                                |
+--------------------------------------------+--------------------------------------------+
                                             |
                                             v
+-----------------------------------------------------------------------------------------+
|                                   POSTGRESQL ENGINE                                     |
|                                                                                         |
|  +-------------------+       +-----------------------+                                  |
|  |   orders Table    |       |  outbox_events Table  |                                  |
|  +-------------------+       +-----------------------+                                  |
|            \                             /                                              |
|             \                           /                                               |
|              v                         v                                                |
|       [ PostgreSQL Write-Ahead Log (WAL Engine - wal_level = logical) ]                 |
|                                     |                                                   |
|                                     v                                                   |
|       [ Logical Decoding Plugin: pgoutput (Replication Slot: debezium_outbox) ]         |
+-------------------------------------+---------------------------------------------------+
                                      |  Binary Streaming Replication Protocol
                                      v
+-----------------------------------------------------------------------------------------+
|                                  KAFKA CONNECT CLUSTER                                  |
|                                                                                         |
|  [ Debezium PostgreSQL Connector ]                                                      |
|         |                                                                               |
|         v                                                                               |
|  [ Outbox Event Router SMT (Single Message Transformation) ]                            |
|    * Unwraps CDC envelope (discards before/after/source metadata)                       |
|    * Extracts payload column -> Kafka message value                                     |
|    * Extracts aggregate_id   -> Kafka message key (preserves partition ordering)        |
|    * Extracts aggregate_type -> Target topic routing (events.Order)                     |
|    * Injects traceparent & event_type into Kafka Record Headers                         |
+-------------------------------------+---------------------------------------------------+
                                      |  TCP Produce (acks=all)
                                      v
+-----------------------------------------------------------------------------------------+
|                                      APACHE KAFKA                                       |
|                                                                                         |
|  Topic: events.Order (Key: ord_123)                                                     |
|  Partition 3 [ Offset 40182 ]: { "eventType": "OrderCreated", "amount": 99.00 }        |
+-------------------------------------+---------------------------------------------------+
                                      |
                                      v
+-----------------------------------------------------------------------------------------+
|                           DOWNSTREAM IDEMPOTENT CONSUMERS                               |
|                                                                                         |
|  BEGIN TRANSACTION;                                                                     |
|    INSERT INTO processed_events (event_id, consumer_group) VALUES ('9c8f...', 'billing')|
|    ON CONFLICT (event_id, consumer_group) DO NOTHING RETURNING event_id;                |
|    -- If insert succeeded: execute billing mutation                                     |
|  COMMIT;                                                                                |
+-----------------------------------------------------------------------------------------+

The Role of Logical Decoding and the pgoutput Plugin

PostgreSQL appends all database modifications to its Write-Ahead Log (WAL) before writing dirty buffer pages to the underlying table heap. When wal_level is set to logical, PostgreSQL writes extra metadata to the WAL segments, including old and new row states and transaction commit boundaries.

The logical decoding engine reads these binary WAL changes and translates them into a discrete logical stream of data changes via a decoding plugin. Historically, third-party plugins like decoderbufs or wal2json were required. Modern enterprise architectures use pgoutput, PostgreSQL's native logical decoding plugin introduced in PostgreSQL 10 (see PostgreSQL Logical Replication Protocol Documentation) and maintained as the core foundation for logical replication.

Because pgoutput is an in-core C library, it eliminates external shared library dependencies, incurs minimal CPU overhead, and streams change events directly to authorized replication clients via the standard PostgreSQL streaming replication protocol.

Decoupling Domain Events from CDC Envelopes via Debezium SMT

A standard Debezium PostgreSQL connector extracts raw CDC events. By default, Debezium produces an envelope containing database internals:

{
  "before": null,
  "after": {
    "id": "9c8fb9ce-1234-4b47-8a6c-7e6d2b5e3f10",
    "aggregate_type": "Order",
    "aggregate_id": "ord_1042",
    "event_type": "OrderCreated",
    "payload": "{\"total\": 149.50, \"currency\": \"USD\"}",
    "created_at": 1729814400000
  },
  "source": { "version": "2.7.0.Final", "connector": "postgresql", "db": "commerce", "ts_ms": 1729814400050, "lsn": 24892180 }
}

Exposing database-level CDC envelopes to downstream business microservices creates tight coupling between internal database schema definitions and external contracts.

Debezium provides the Outbox Event Router Single Message Transformation (SMT) (io.debezium.transforms.outbox.EventRouter). The SMT intercepts the change stream directly within the Kafka Connect worker process before serialization and transforms it into a pure domain event:

1. Payload Extraction: It unwraps the payload column and places it directly into the Kafka record's value payload.

2. Key Assignment: It sets the Kafka message key to the aggregate_id value, ensuring all events for the same business entity hash to the same Kafka partition, preserving strict chronological ordering.

3. Topic Routing: It extracts the aggregate_type column and dynamically routes the message to a specific topic pattern (e.g., events.Order).

4. Header Ingestion: It extracts contextual fields—such as event_type, trace identifiers, and schema versions—and writes them as native Kafka record headers.

---

3. Production PostgreSQL Outbox DDL and Declarative Partitioning

A high-volume outbox table in an enterprise system processes tens of millions of records weekly. A flat, unpartitioned table subject to recurring row deletions causes catastrophic page fragmentation, WAL amplification, and autovacuum degradation.

The schema below utilizes PostgreSQL native declarative range partitioning based on created_at.

-- Step 1: Ensure required extensions are available
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- Step 2: Create the partitioned outbox table
-- Range partitioned by created_at to support zero-cost drop-partition retention
CREATE TABLE public.outbox_events (
    created_at       TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
    id               UUID NOT NULL DEFAULT uuid_generate_v4(),
    aggregate_type   VARCHAR(128) NOT NULL,
    aggregate_id     VARCHAR(128) NOT NULL,
    event_type       VARCHAR(128) NOT NULL,
    payload          JSONB NOT NULL,
    headers          JSONB DEFAULT '{}'::jsonb,
    CONSTRAINT pk_outbox_events PRIMARY KEY (created_at, id)
) PARTITION BY RANGE (created_at);

-- Step 3: Set Replica Identity
-- For append-only outbox tables where rows are never modified in-place,
-- DEFAULT replica identity records the primary key in the WAL, saving bandwidth.
ALTER TABLE public.outbox_events REPLICA IDENTITY DEFAULT;

-- Step 4: Create performance indexes
-- Local composite index on partitions for aggregate timeline tracing
CREATE INDEX idx_outbox_aggregate 
ON public.outbox_events (aggregate_type, aggregate_id, created_at DESC);

-- Step 5: Establish initial daily partitions
-- Daily partitioning isolates writes to the active day's partition
CREATE TABLE public.outbox_events_default 
    PARTITION OF public.outbox_events DEFAULT;

CREATE TABLE public.outbox_events_2026_10_26 
    PARTITION OF public.outbox_events 
    FOR VALUES FROM ('2026-10-26 00:00:00+00') TO ('2026-10-27 00:00:00+00');

CREATE TABLE public.outbox_events_2026_10_27 
    PARTITION OF public.outbox_events 
    FOR VALUES FROM ('2026-10-27 00:00:00+00') TO ('2026-10-28 00:00:00+00');

CREATE TABLE public.outbox_events_2026_10_28 
    PARTITION OF public.outbox_events 
    FOR VALUES FROM ('2026-10-28 00:00:00+00') TO ('2026-10-29 00:00:00+00');

-- Step 6: Dedicated heartbeat table to advance LSN during low-volume periods
CREATE TABLE public.debezium_heartbeat (
    id          INTEGER PRIMARY KEY,
    last_beat   TIMESTAMPTZ NOT NULL
);

INSERT INTO public.debezium_heartbeat (id, last_beat) 
VALUES (1, clock_timestamp())
ON CONFLICT (id) DO NOTHING;

Writing Business State and Outbox Records Atomically

In the application persistence tier, both the entity modification and the outbox event insert must execute within the exact same transaction block.

BEGIN;

-- 1. Mutate primary business entity
INSERT INTO public.orders (
    id, 
    customer_id, 
    status, 
    total_amount, 
    created_at
) VALUES (
    'ord_84920194', 
    'cust_99214', 
    'SUBMITTED', 
    249.99, 
    clock_timestamp()
);

-- 2. Insert outbox event record
INSERT INTO public.outbox_events (
    id,
    aggregate_type,
    aggregate_id,
    event_type,
    payload,
    headers,
    created_at
) VALUES (
    'a3c74f51-382a-41e9-9159-86dbbb31cf42',
    'Order',
    'ord_84920194',
    'OrderSubmitted',
    jsonb_build_object(
        'orderId', 'ord_84920194',
        'customerId', 'cust_99214',
        'totalAmount', 249.99,
        'currency', 'USD',
        'itemCount', 3
    ),
    jsonb_build_object(
        'traceparent', '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01',
        'schemaVersion', 'v1'
    ),
    clock_timestamp()
);

COMMIT;

---

4. Production Kafka Connect and Debezium SMT Configuration

Below is the complete, production-grade JSON configuration payload submitted to the Kafka Connect REST API (PUT /connectors/debezium-postgresql-outbox/config).

This configuration establishes the connection, attaches to the pgoutput replication slot, enables Outbox Event Router SMT, configures heartbeats to advance the Log Sequence Number (LSN), and serializes message payloads as raw JSON.

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "tasks.max": "1",
  "plugin.name": "pgoutput",
  "database.hostname": "postgres-primary.internal.network",
  "database.port": "5432",
  "database.user": "debezium_cdc_user",
  "database.password": "${file:/secrets/db-credentials.properties:cdc_password}",
  "database.dbname": "commerce_db",
  "database.server.name": "commerce_cdc",
  "topic.prefix": "commerce_cdc",
  
  "slot.name": "debezium_outbox_slot",
  "slot.drop.on.stop": "false",
  "publication.name": "debezium_outbox_publication",
  "publication.autocreate.mode": "filtered",
  
  "table.include.list": "public.outbox_events,public.debezium_heartbeat",
  "tombstones.on.delete": "false",
  "decimal.handling.mode": "double",
  "money.handling.mode": "precise",
  
  "heartbeat.interval.ms": "5000",
  "heartbeat.action.query": "INSERT INTO public.debezium_heartbeat (id, last_beat) VALUES (1, clock_timestamp()) ON CONFLICT (id) DO UPDATE SET last_beat = EXCLUDED.last_beat;",
  
  "transforms": "outbox",
  "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
  "transforms.outbox.route.by.field": "aggregate_type",
  "transforms.outbox.route.topic.replacement": "events.${routedByValue}",
  "transforms.outbox.table.field.event.id": "id",
  "transforms.outbox.table.field.event.key": "aggregate_id",
  "transforms.outbox.table.field.event.type": "event_type",
  "transforms.outbox.table.field.event.payload": "payload",
  "transforms.outbox.table.field.event.timestamp": "created_at",
  "transforms.outbox.table.fields.additional.placement": "headers:header:metadata,event_type:header:eventType",
  "transforms.outbox.tracing.with.context.field.header": "traceparent",
  
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter.schemas.enable": "false",
  
  "max.batch.size": "4096",
  "max.queue.size": "16384",
  "poll.interval.ms": "100"
}

Technical Parameter Breakdown

---

5. Vacuuming, Partition Pruning, and WAL Bloat Mitigation

In an outbox deployment handling sustained write volume, failure to manage PostgreSQL storage internals will lead to service interruption. Two interrelated storage hazards must be mitigated: Replication Slot WAL Retainage and Table Heap Bloat.

The Replication Slot WAL Retention Trap

PostgreSQL guarantees that no transaction log segment required by an active replication slot is recycled or removed. PostgreSQL tracks the consumer's progress via confirmed_flush_lsn.

If the Kafka Connect cluster crashes, experiences an unhandled network partition, or stalls during a deployment:

1. Debezium ceases acknowledging LSN positions to PostgreSQL.

2. PostgreSQL halts WAL segment deletion in the pg_wal directory.

3. Every subsequent database mutation writes new 16MB WAL segments to disk.

4. If unchecked, the host storage volume reaches 100% capacity, forcing PostgreSQL into an immediate emergency read-only shutdown.

Protective Configuration: max_slot_wal_keep_size

To defend the primary database cluster from complete disk exhaustion caused by an unresponsive replication consumer, configure max_slot_wal_keep_size in postgresql.conf:

; Cap maximum WAL retention for any single replication slot to 64GB
max_slot_wal_keep_size = 65536MB

; Increase replication timeout sensitivity (default 60s)
wal_sender_timeout = 30s

If Debezium disconnects and replication lag exceeds 64GB, PostgreSQL intentionally invalidates the replication slot (wal_status = 'lost'), drops the retained WAL segments, and preserves database availability. While Debezium will require a managed re-snapshot or backfill, primary database operations remain completely uninterrupted.

Continuous Monitoring Query for Replication Lag

Execute this query via Prometheus PostgreSQL Exporter or Datadog agent at 15-second intervals:

SELECT
    slot_name,
    plugin,
    active,
    wal_status,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS replication_lag_bytes,
    ROUND(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) / 1024 / 1024, 2) AS lag_mb
FROM pg_replication_slots
WHERE slot_name = 'debezium_outbox_slot';

Alert immediately if lag_mb exceeds 10,000MB (10GB) or if active registers false for more than 120 seconds.

The High-Volume DELETE Anti-Pattern vs Partition Dropping

A common architectural error in early outbox implementations is using a background cron job to execute row-level cleanups:

-- DANGEROUS AT SCALE: Do not execute in high-throughput production
DELETE FROM outbox_events WHERE created_at < NOW() - INTERVAL '48 hours';

In PostgreSQL's Multi-Version Concurrency Control (MVCC) architecture, an SQL DELETE does not remove bytes from disk. Instead, it marks existing tuples as dead. In an application generating high outbox write volumes:

• Continuous row deletions over multiple days accumulate hundreds of millions of dead tuples on disk.
• Autovacuum must traverse substantial volumes of table heap and B-tree index pages to mark dead space reusable.
• Table indexes bloat significantly, evicting the primary application working set from the operating system page cache.
• The DELETE operations themselves generate millions of new WAL records, compounding disk I/O load.

The Zero-Cost Solution: Declarative Partition Dropping

By range-partitioning the outbox table by day, cleaning up historical events is an instantaneous metadata modification:

-- Drops the entire physical partition instantly
-- 0 dead tuples generated, 0 index pages traversed, 0 WAL bloat
DROP TABLE public.outbox_events_2026_10_25;

Automated Partition Maintenance Function

Deploy the following maintenance stored procedure, executed daily via pg_cron or an external orchestrator:

CREATE OR REPLACE PROCEDURE public.rotate_outbox_partitions(
    retention_days INTEGER DEFAULT 3,
    pre_create_days INTEGER DEFAULT 2
)
LANGUAGE plpgsql
AS $$
DECLARE
    target_date DATE;
    partition_name TEXT;
    start_ts TIMESTAMPTZ;
    end_ts TIMESTAMPTZ;
    sql_stmt TEXT;
BEGIN
    -- 1. Pre-create forward partitions to prevent routing into the default partition
    FOR i IN 0..pre_create_days LOOP
        target_date := CURRENT_DATE + i;
        partition_name := 'outbox_events_' || to_char(target_date, 'YYYY_MM_DD');
        start_ts := target_date::TIMESTAMPTZ;
        end_ts := (target_date + 1)::TIMESTAMPTZ;

        IF NOT EXISTS (
            SELECT 1 FROM pg_class c 
            JOIN pg_namespace n ON n.oid = c.relnamespace 
            WHERE c.relname = partition_name AND n.nspname = 'public'
        ) THEN
            sql_stmt := format(
                'CREATE TABLE public.%I PARTITION OF public.outbox_events FOR VALUES FROM (%L) TO (%L);',
                partition_name, start_ts, end_ts
            );
            EXECUTE sql_stmt;
            RAISE NOTICE 'Created outbox partition: %', partition_name;
        END IF;
    END LOOP;

    -- 2. Drop expired historical partitions older than the retention boundary
    FOR partition_name IN
        SELECT c.relname
        FROM pg_class c
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE n.nspname = 'public'
          AND c.relname ~ '^outbox_events_[0-9]{4}_[0-9]{2}_[0-9]{2}$'
          AND to_date(substring(c.relname from 'outbox_events_([0-9_]+)'), 'YYYY_MM_DD') 
              < (CURRENT_DATE - retention_days)
    LOOP
        sql_stmt := format('DROP TABLE public.%I;', partition_name);
        EXECUTE sql_stmt;
        RAISE NOTICE 'Dropped expired outbox partition: %', partition_name;
    END LOOP;
END;
$$;

---

6. End-to-End Idempotent Consumer Implementation in TypeScript

Because Debezium, the streaming replication protocol, and Apache Kafka adhere to at-least-once delivery guarantees, network retransmissions, broker restarts, and consumer group rebalances will occasionally deliver duplicate events to downstream consumers.

Downstream consumers must be architected as strictly idempotent state machines.

                           [ Kafka Topic: events.Order ]
                                         |
                                         v
                         [ Consumer: eachMessage Batch ]
                                         |
                                         v
                       [ Start PostgreSQL Transaction ]
                                         |
                                         v
             [ INSERT INTO processed_events ON CONFLICT DO NOTHING ]
                                         |
                       +-----------------+-----------------+
                       |                                   |
              (Row Inserted: Novel)              (Conflict: Duplicate)
                       |                                   |
                       v                                   v
             [ Execute Business Logic ]           [ Log Duplicate Notice ]
             [ Update Inventory / Balances ]               |
                       |                                   |
                       +-----------------+-----------------+
                                         |
                                         v
                            [ COMMIT SQL TRANSACTION ]
                                         |
                                         v
                            [ Commit Kafka Consumer Offset ]

PostgreSQL Deduplication Store DDL

CREATE TABLE public.processed_events (
    event_id        UUID NOT NULL,
    consumer_group  VARCHAR(128) NOT NULL,
    processed_at    TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
    CONSTRAINT pk_processed_events PRIMARY KEY (event_id, consumer_group)
);

-- Optimize automatic retention for deduplication records
CREATE INDEX idx_processed_events_timestamp 
ON public.processed_events (processed_at);

Complete Production TypeScript Consumer

Below is an enterprise-grade TypeScript consumer implementing deterministic deduplication, atomic transaction isolation, and offset management using kafkajs and pg.

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';
import { Pool, PoolClient } from 'pg';

interface OrderSubmittedEvent {
  orderId: string;
  customerId: string;
  totalAmount: number;
  currency: string;
  itemCount: number;
}

export class IdempotentOrderConsumer {
  private kafkaConsumer: Consumer;
  private dbPool: Pool;
  private readonly consumerGroupId = 'fulfillment-order-service';

  constructor(kafkaBrokers: string[], databaseUrl: string) {
    const kafka = new Kafka({
      clientId: 'fulfillment-service-consumer',
      brokers: kafkaBrokers,
      retry: {
        initialRetryTime: 300,
        retries: 8,
      },
    });

    this.kafkaConsumer = kafka.consumer({
      groupId: this.consumerGroupId,
      sessionTimeout: 30000,
      heartbeatInterval: 3000,
    });

    this.dbPool = new Pool({
      connectionString: databaseUrl,
      max: 20,
      idleTimeoutMillis: 10000,
    });
  }

  public async start(): Promise<void> {
    await this.kafkaConsumer.connect();
    await this.kafkaConsumer.subscribe({
      topic: 'events.Order',
      fromBeginning: false,
    });

    await this.kafkaConsumer.run({
      autoCommit: false, // Explicit manual offset commit after DB transaction
      eachMessage: async (payload: EachMessagePayload) => {
        await this.handleMessageWithDeduplication(payload);
      },
    });
  }

  private async handleMessageWithDeduplication({
    topic,
    partition,
    message,
  }: EachMessagePayload): Promise<void> {
    if (!message.value) {
      return;
    }

    // Extract headers propagated by Debezium Outbox SMT
    const eventTypeHeader = message.headers?.eventType?.toString();
    const metadataHeader = message.headers?.metadata?.toString();
    const metadata = metadataHeader ? JSON.parse(metadataHeader) : {};

    // In Debezium SMT, event id is carried in metadata or can be parsed from payload
    const parsedPayload: OrderSubmittedEvent = JSON.parse(message.value.toString());
    const eventId = metadata.eventId || message.key?.toString() || parsedPayload.orderId;

    if (!eventId) {
      throw new Error(`Message at partition ${partition} offset ${message.offset} lacks deterministic ID`);
    }

    const client: PoolClient = await this.dbPool.connect();

    try {
      await client.query('BEGIN;');

      // Atomic Deduplication Gate:
      // Try to register event ID for this specific consumer group
      const dedupQuery = `
        INSERT INTO public.processed_events (event_id, consumer_group, processed_at)
        VALUES ($1, $2, clock_timestamp())
        ON CONFLICT (event_id, consumer_group) DO NOTHING
        RETURNING event_id;
      `;
      const dedupResult = await client.query(dedupQuery, [eventId, this.consumerGroupId]);

      if (dedupResult.rowCount === 0) {
        // Conflict detected: event was already successfully processed
        await client.query('COMMIT;');
        
        // Acknowledge offset immediately to advance Kafka consumer past the duplicate
        await this.kafkaConsumer.commitOffsets([
          {
            topic,
            partition,
            offset: (BigInt(message.offset) + 1n).toString(),
          },
        ]);
        return;
      }

      // Novel Event: Execute business domain operations within transaction boundary
      await this.executeDomainOperation(client, eventTypeHeader, parsedPayload);

      // Commit business transaction and deduplication record together
      await client.query('COMMIT;');

      // Commit Kafka offset only after the database guarantees durability
      await this.kafkaConsumer.commitOffsets([
        {
          topic,
          partition,
          offset: (BigInt(message.offset) + 1n).toString(),
        },
      ]);
    } catch (error) {
      await client.query('ROLLBACK;');
      // Re-throw error to trigger KafkaJS backoff and retry mechanism
      throw error;
    } finally {
      client.release();
    }
  }

  private async executeDomainOperation(
    client: PoolClient,
    eventType: string | undefined,
    data: OrderSubmittedEvent
  ): Promise<void> {
    if (eventType !== 'OrderSubmitted') {
      return;
    }

    const insertFulfillmentSql = `
      INSERT INTO public.fulfillment_jobs (
        order_id, customer_id, total_units, status, scheduled_at
      ) VALUES ($1, $2, $3, 'QUEUED', clock_timestamp());
    `;
    await client.query(insertFulfillmentSql, [data.orderId, data.customerId, data.itemCount]);
  }

  public async shutdown(): Promise<void> {
    await this.kafkaConsumer.disconnect();
    await this.dbPool.end();
  }
}

---

7. Architectural Trade-Offs and Technology Comparison

Selecting an event integration pattern requires balancing consistency requirements against operational overhead.

Enterprise Implementation Considerations

While the Transactional Outbox pattern with Debezium and Kafka provides an industry standard for data consistency, implementing it in enterprise systems requires specialized operational discipline: configuring replication slots, tuning PostgreSQL WAL write parameters, defining automated partition policies, and enforcing strict idempotency contracts across microservices.

Organizations scaling mission-critical platforms or modernizing distributed legacy stacks frequently engage Wise Hustlers custom software engineering to architect fault-tolerant event backbones, optimize relational persistence layers, and eliminate distributed consistency anomalies.

---

8. Frequently Asked Questions

What happens to Debezium's logical replication slot during a PostgreSQL failover?

In standard PostgreSQL streaming replication architectures (including Patroni, AWS RDS Multi-AZ, and Google Cloud SQL), logical replication slots are maintained only on the primary instance. When failover occurs and a standby is promoted to primary, standard replication slots do not automatically fail over. In PostgreSQL 17, native logical replication failover slots (failover = true on the replication slot) permit replication slots to synchronize continuously to physical standbys. For PostgreSQL 14 through 16 clusters, enterprise teams use utilities such as pg_failover_slots or Patroni's slot synchronization mechanics to ensure the standby possesses a matched LSN slot before promotion, enabling Debezium to reconnect without missing events or re-snapshotting the database.

Why use the Outbox Event Router SMT instead of writing events directly to destination Kafka topics in application code?

Writing directly to Kafka from application code introduces the dual-write failure modes detailed in Section 1. If the database commit succeeds but the Kafka producer times out, the event is permanently lost. If the Kafka produce succeeds but the database rolls back, phantom events trigger illegitimate downstream side effects. The Outbox Event Router SMT bridges this divide by extracting records committed atomically to the PostgreSQL outbox table and routing them directly to domain-specific Kafka topics based on table column values, preserving strict transactional atomicity.

How do we prevent PostgreSQL disk exhaustion if the Kafka Connect cluster crashes?

Set max_slot_wal_keep_size in postgresql.conf (for example, 64GB). If Kafka Connect fails and replication lag exceeds this threshold, PostgreSQL intentionally invalidates the replication slot rather than filling the host disk volume to 100%. While this requires re-initializing the Debezium connector following prolonged outages, it protects the primary database cluster from complete operational lockup. Continuous Prometheus alerting on pg_replication_slots.replication_lag_bytes should notify platform teams long before this threshold is breached.

Can we use PostgreSQL LISTEN and NOTIFY instead of Debezium CDC for the Outbox Pattern?

No, PostgreSQL LISTEN and NOTIFY are unsuitable for durable, high-throughput enterprise event backbones. First, NOTIFY carries a hard payload ceiling of 8,000 bytes, which cannot accommodate enterprise domain schemas. Second, NOTIFY queues are ephemeral in memory: if a listening process disconnects or restarts, all messages emitted during the downtime are permanently lost without recovery. Debezium CDC tails the persistent Write-Ahead Log on disk, providing durable at-least-once delivery with complete offset recovery across arbitrary system downtime.

---

Sources

• Debezium Documentation: Outbox Event Router Single Message Transformation
• PostgreSQL Documentation: Chapter 49 - Logical Replication and pgoutput
• PostgreSQL Documentation: Chapter 27 - High Availability, Load Balancing, and Replication Slots
• PostgreSQL 10 Release Notes: Native Logical Replication and pgoutput
• Patroni Documentation: Permanent Logical Replication Slots Synchronization
• EnterpriseDB: pg_failover_slots High-Availability Extension
• Apache Kafka Documentation: Message Delivery Semantics and Exactly-Once Processing
• Martin Kleppmann: Designing Data-Intensive Applications (O'Reilly Media)
• Chris Richardson: Pattern: Transactional Outbox (Microservices.io)