Practical application of Kafka Outbox Pattern

Introduction

The most frequently broken assumption in event-based architecture is the expectation that “DB commit and message publication will succeed simultaneously.” In actual operation, there will always be cases where only one of the two succeeds. The Outbox pattern is a realistic way to bridge this gap.

This article summarizes schema design, relay process, and consumer idempotency required when introducing Outbox in a Kafka-based service.

Problem definition

If operated without Outbox, the following problems will occur again.

• The order DB is successful, but the downstream system is not updated due to event publishing failure.
• The event is issued, but a DB rollback occurs, resulting in a “ghost event”.
• During the retry process, the same event is consumed multiple times, resulting in an incorrect state.
• There is no standard for when to start reprocessing when recovering from a failure.

The key is to atomically process “state change + event recording” within a DB transaction.

Key concepts

Code example 1: Outbox recording within a transaction

import { db } from "@/lib/db";

export async function placeOrder(command: {
  orderId: string;
  userId: string;
  amount: number;
}) {
  await db.transaction(async (tx) => {
    await tx.query(
      `INSERT INTO orders (id, user_id, amount, status)
       VALUES ($1, $2, $3, 'PLACED')`,
      [command.orderId, command.userId, command.amount],
    );

    await tx.query(
      `INSERT INTO outbox_events (event_id, topic, aggregate_id, payload, status)
       VALUES ($1, 'order.placed', $2, $3::jsonb, 'PENDING')`,
      [
        crypto.randomUUID(),
        command.orderId,
        JSON.stringify({
          orderId: command.orderId,
          userId: command.userId,
          amount: command.amount,
        }),
      ],
    );
  });
}

Code example 2: Relay Worker (batch publishing + state transition)

export async function relayOutboxBatch(limit = 100) {
  const rows = await db.query(
    `SELECT event_id, topic, payload
     FROM outbox_events
     WHERE status = 'PENDING'
     ORDER BY created_at ASC
     LIMIT $1
     FOR UPDATE SKIP LOCKED`,
    [limit],
  );

  for (const row of rows) {
    try {
      await kafkaProducer.send({
        topic: row.topic,
        messages: [{ key: row.event_id, value: JSON.stringify(row.payload) }],
      });

      await db.query(
        `UPDATE outbox_events SET status = 'SENT', sent_at = now() WHERE event_id = $1`,
        [row.event_id],
      );
    } catch {
      await db.query(
        `UPDATE outbox_events
         SET status = 'FAILED', retry_count = retry_count + 1
         WHERE event_id = $1`,
        [row.event_id],
      );
    }
  }
}

Architecture flow

Tradeoffs

• Outbox improves data consistency, but adds table/worker operating costs.
• If near real-time is important, tuning the relay cycle and batch size is necessary.
• If consumer idempotence is not designed together, the problem of duplicate events will remain.

Cleanup

The Outbox pattern is a strategy that is stronger in imperfect reality than in perfect theory. By tying state changes and event records to the same transaction and managing relays/consumers on an operational basis, the likelihood of recovery in the event of a failure is greatly increased.

Image source

• Cover: source link
• License: CC BY-SA 3.0 / Author: BalticServers.com
• Note: After downloading the free license image from Wikimedia Commons, it was optimized to JPG at 1600px.