Fault-Tolerant Payment Pipeline

Build a distributed payment pipeline with outbox, idempotency, and tracing.

Introduction

30 Second Summary

Every time you tap "Pay" on your phone, a chain of services must coordinate without losing your money. If any link in that chain fails, the system needs to recover gracefully without charging you twice or losing the transaction entirely.

In this project, you will build a multi-service payment processing pipeline that guarantees zero message loss, processes each event exactly once, and traces every request end-to-end across async boundaries.

What You'll Build

A three-service TypeScript pipeline communicating via Kafka (Redpanda), with PostgreSQL for persistence, circuit breaker protection for fault isolation, and OpenTelemetry distributed tracing visualized in Jaeger.

By the end of this project, you'll have:

  • A Docker Compose stack running Redpanda, PostgreSQL, and Jaeger locally with a single command.
  • An Order API that atomically writes orders and outbox events in a single database transaction, guaranteeing zero message loss via the transactional outbox pattern.
  • An idempotent Payment Consumer protected by a circuit breaker that gracefully degrades under failure and routes poisoned messages to a dead letter queue.
  • Secret Mission: Add Prometheus metrics and a Grafana dashboard showing circuit breaker state transitions and message throughput in real time.

Are there any prerequisites?

This project assumes you are comfortable writing TypeScript and familiar with building REST APIs. You should also have basic Docker knowledge (pulling images, running containers).

No prior experience with event-driven architecture, Kafka, or distributed tracing is required.

Set Up the Project

Before writing any application code, you need a solid foundation. This step creates the project structure, installs every dependency, and launches the infrastructure services your pipeline relies on.

By the end of this step you will have a TypeScript project with all packages ready, plus a local Docker stack running a message broker, a database, and a tracing backend.

In this step, get ready to:
  • Create the project folder, package.json, and tsconfig.json, then install all npm dependencies.
  • Define a Docker Compose stack with Redpanda, PostgreSQL, and Jaeger.
  • Create the database schema and start the entire infrastructure.
Create project structure and install dependencies

Every service in the pipeline shares the same dependencies and TypeScript configuration. A single package.json at the project root keeps things simple.

  • Open a terminal by pressing the Windows key, typing PowerShell, and pressing Enter.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir payment-pipeline
cd payment-pipeline
  • Open the project in VS Code by running:
code .

VS Code opens with the payment-pipeline folder shown in the Explorer sidebar.

code command not recognized?

If your terminal shows an error, VS Code may not be in your system PATH. Close and reopen your terminal, then try again.

If that does not help, open VS Code manually (press the Windows key, type VS Code, press Enter). Then click File, then Open Folder, and select payment-pipeline on your Desktop.

Help me add VS Code to my PATH on Windows

  • In VS Code, click the New File icon at the top of the Explorer sidebar (the small page icon with a plus sign).
  • Name the file package.json and press Enter.
  • Paste the following content into package.json:
{
  "name": "payment-pipeline",
  "version": "1.0.0",
  "private": true,
  "scripts": {
    "order-api": "npx tsx src/order-api/index.ts",
    "outbox-poller": "npx tsx src/outbox-poller/index.ts",
    "payment-consumer": "npx tsx src/payment-consumer/index.ts"
  },
  "dependencies": {
    "express": "4.19.2",
    "kafkajs": "2.2.4",
    "opossum": "10.0.0",
    "pg": "8.22.0",
    "@opentelemetry/api": "1.9.1",
    "@opentelemetry/sdk-node": "0.219.0",
    "@opentelemetry/sdk-trace-node": "2.8.0",
    "@opentelemetry/exporter-trace-otlp-proto": "0.219.0",
    "@opentelemetry/instrumentation-http": "0.219.0",
    "@opentelemetry/instrumentation-pg": "0.70.0",
    "@opentelemetry/resources": "2.8.0",
    "@opentelemetry/semantic-conventions": "1.29.0"
  },
  "devDependencies": {
    "@types/express": "5.0.6",
    "@types/pg": "8.20.0",
    "@types/opossum": "8.1.7",
    "tsx": "4.19.0",
    "typescript": "5.8.3"
  }
}

What does this file do?

  • The scripts section defines how to start each of the three services using tsx (a TypeScript runner that needs no separate compile step).
  • The dependencies include Express for the API, KafkaJS for messaging, opossum for circuit breaking, pg for PostgreSQL, and the OpenTelemetry SDK for distributed tracing.
  • The devDependencies provide TypeScript type definitions and the tsx runtime itself.
  • Save package.json by pressing Ctrl+S.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

{
  "name": "payment-pipeline",
  "version": "1.0.0",
  "private": true,
  "scripts": {
    "order-api": "npx tsx src/order-api/index.ts",
    "outbox-poller": "npx tsx src/outbox-poller/index.ts",
    "payment-consumer": "npx tsx src/payment-consumer/index.ts"
  },
  "dependencies": {
    "express": "4.19.2",
    "kafkajs": "2.2.4",
    "opossum": "10.0.0",
    "pg": "8.22.0",
    "@opentelemetry/api": "1.9.1",
    "@opentelemetry/sdk-node": "0.219.0",
    "@opentelemetry/sdk-trace-node": "2.8.0",
    "@opentelemetry/exporter-trace-otlp-proto": "0.219.0",
    "@opentelemetry/instrumentation-http": "0.219.0",
    "@opentelemetry/instrumentation-pg": "0.70.0",
    "@opentelemetry/resources": "2.8.0",
    "@opentelemetry/semantic-conventions": "1.29.0"
  },
  "devDependencies": {
    "@types/express": "5.0.6",
    "@types/pg": "8.20.0",
    "@types/opossum": "8.1.7",
    "tsx": "4.19.0",
    "typescript": "5.8.3"
  }
}
  • Create another new file in the Explorer sidebar and name it tsconfig.json.
  • Paste the following content into tsconfig.json:
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "ESNext",
    "moduleResolution": "bundler",
    "esModuleInterop": true,
    "strict": true,
    "outDir": "dist",
    "rootDir": "src",
    "skipLibCheck": true
  },
  "include": ["src"]
}

What does this configure?

This tells TypeScript to compile to modern JavaScript (ES2022), use ESNext modules resolved with the bundler strategy, and enforce strict type checking. The skipLibCheck flag speeds up compilation by skipping type checks on third-party declaration files.

  • Save tsconfig.json by pressing Ctrl+S.

✔️ Awesome, I've got everything!

Great. You now have both config files saved.

ⓧ I'd like to double check the full code

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "ESNext",
    "moduleResolution": "bundler",
    "esModuleInterop": true,
    "strict": true,
    "outDir": "dist",
    "rootDir": "src",
    "skipLibCheck": true
  },
  "include": ["src"]
}

Now install all the packages declared in package.json.

  • Open VS Code's integrated terminal by pressing Ctrl+` (backtick key, usually above Tab).
  • Install all dependencies by running:
npm install

This takes a minute to download all packages. When it finishes, you should see a node_modules folder appear in the Explorer sidebar and a package-lock.json file next to your package.json.

npm install failing?

  • Make sure your terminal's working directory is inside payment-pipeline. Run pwd to confirm. If you are somewhere else, run cd ~/Desktop/payment-pipeline.
  • Check your internet connection. npm needs to download packages from the registry.
  • If you see permission errors, try closing and reopening your terminal as Administrator.

Help me fix npm install errors

Create the Docker Compose stack

Your pipeline needs three infrastructure services. Redpanda is a Kafka-compatible message broker that starts in seconds with minimal memory. PostgreSQL stores orders and the outbox table. Jaeger collects distributed traces so you can visualize the entire event flow.

Why Redpanda instead of Apache Kafka?

Redpanda is a single binary with no ZooKeeper dependency. It uses about 512 MB of memory compared to Kafka's 1.5 GB+, and starts in a few seconds. It speaks the same Kafka protocol, so your KafkaJS code works identically against either.

  • Create a new file in the Explorer sidebar and name it docker-compose.yml.
  • Paste the following content into docker-compose.yml:
services:
  redpanda:
    image: redpandadata/redpanda:latest
    command:
      - redpanda start
      - --smp 1
      - --memory 512M
      - --overprovisioned
      - --node-id 0
      - --kafka-addr PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092
      - --advertise-kafka-addr PLAINTEXT://redpanda:29092,OUTSIDE://localhost:9092
    ports:
      - "9092:9092"
      - "29092:29092"

  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: pipeline
      POSTGRES_PASSWORD: pipeline
      POSTGRES_DB: payments
    ports:
      - "5432:5432"
    volumes:
      - ./migrations/init.sql:/docker-entrypoint-initdb.d/init.sql

  jaeger:
    image: jaegertracing/all-in-one:latest
    environment:
      COLLECTOR_OTLP_ENABLED: "true"
    ports:
      - "16686:16686"
      - "4318:4318"

What does each service do?

  • redpanda exposes port 9092 for external Kafka clients (your Node.js services) and 29092 for internal container-to-container traffic.
  • postgres creates a database called payments with user pipeline. The volume mount tells it to run any SQL files in ./migrations/ on first startup.
  • jaeger accepts OpenTelemetry traces on port 4318 (OTLP/HTTP) and serves a web UI on port 16686.
  • Save docker-compose.yml by pressing Ctrl+S.

✔️ Awesome, I've got everything!

Great. Do not start the containers yet. You need the migration file first so PostgreSQL can initialize the schema on first boot.

ⓧ I'd like to double check the full code

services:
  redpanda:
    image: redpandadata/redpanda:latest
    command:
      - redpanda start
      - --smp 1
      - --memory 512M
      - --overprovisioned
      - --node-id 0
      - --kafka-addr PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092
      - --advertise-kafka-addr PLAINTEXT://redpanda:29092,OUTSIDE://localhost:9092
    ports:
      - "9092:9092"
      - "29092:29092"

  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: pipeline
      POSTGRES_PASSWORD: pipeline
      POSTGRES_DB: payments
    ports:
      - "5432:5432"
    volumes:
      - ./migrations/init.sql:/docker-entrypoint-initdb.d/init.sql

  jaeger:
    image: jaegertracing/all-in-one:latest
    environment:
      COLLECTOR_OTLP_ENABLED: "true"
    ports:
      - "16686:16686"
      - "4318:4318"
Create the database schema and start the stack

The PostgreSQL container automatically runs any SQL files mounted into its /docker-entrypoint-initdb.d/ directory the first time it starts. You need to create that file before booting the containers.

  • In the Explorer sidebar, click the New Folder icon and name it migrations.
  • Inside the migrations folder, create a new file called init.sql.
  • Paste the following SQL into migrations/init.sql:
CREATE TABLE orders (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id VARCHAR(255) NOT NULL,
  amount DECIMAL(10, 2) NOT NULL,
  currency VARCHAR(3) NOT NULL DEFAULT 'USD',
  status VARCHAR(50) NOT NULL DEFAULT 'pending',
  created_at TIMESTAMP NOT NULL DEFAULT NOW()
);

CREATE TABLE outbox (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_id UUID NOT NULL,
  event_type VARCHAR(100) NOT NULL,
  payload JSONB NOT NULL,
  status VARCHAR(20) NOT NULL DEFAULT 'pending',
  retries INT NOT NULL DEFAULT 0,
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  published_at TIMESTAMP
);

CREATE INDEX idx_outbox_status ON outbox(status) WHERE status = 'pending';

CREATE TABLE processed_events (
  event_id UUID PRIMARY KEY,
  processed_at TIMESTAMP NOT NULL DEFAULT NOW()
);

What do these tables do?

  • orders stores every payment request with its status (pending, paid, etc.).
  • outbox is the heart of the transactional outbox pattern. It holds events that must be published to Kafka. The partial index on status = 'pending' makes polling fast because only unpublished rows are scanned.
  • processed_events records which event IDs the consumer has already handled. This is what makes the consumer idempotent. If it processes the same message twice, the duplicate is detected and skipped.
  • Save migrations/init.sql by pressing Ctrl+S.

✔️ Awesome, I've got everything!

Great. You are ready to start the infrastructure stack.

ⓧ I'd like to double check the full code

CREATE TABLE orders (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id VARCHAR(255) NOT NULL,
  amount DECIMAL(10, 2) NOT NULL,
  currency VARCHAR(3) NOT NULL DEFAULT 'USD',
  status VARCHAR(50) NOT NULL DEFAULT 'pending',
  created_at TIMESTAMP NOT NULL DEFAULT NOW()
);

CREATE TABLE outbox (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_id UUID NOT NULL,
  event_type VARCHAR(100) NOT NULL,
  payload JSONB NOT NULL,
  status VARCHAR(20) NOT NULL DEFAULT 'pending',
  retries INT NOT NULL DEFAULT 0,
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  published_at TIMESTAMP
);

CREATE INDEX idx_outbox_status ON outbox(status) WHERE status = 'pending';

CREATE TABLE processed_events (
  event_id UUID PRIMARY KEY,
  processed_at TIMESTAMP NOT NULL DEFAULT NOW()
);

Now that all files are in place, start the Docker stack.

  • Make sure Docker Desktop is running (look for the whale icon in your system tray).
  • In the VS Code terminal, start all containers by running:
docker compose up -d

Docker pulls the images on first run, which may take a few minutes. When it finishes, you should see each container reported as Started.

  • Verify all containers are healthy by running:
docker compose ps

You should see three services (redpanda, postgres, jaeger) all showing a running state.

  • Confirm the database schema loaded by connecting to PostgreSQL:
docker exec -it payment-pipeline-postgres-1 psql -U pipeline -d payments -c "\dt"

You should see a table listing that includes orders, outbox, and processed_events.

Containers not starting?

  • If you see a port conflict error, another process is already using port 5432 or 9092. Stop the conflicting process or change the host port in docker-compose.yml.
  • If Docker cannot pull images, check your internet connection and that Docker Desktop is fully started (the whale icon should be steady, not animating).
  • If the \dt command returns no tables, the init script did not run. Run docker compose down -v to remove volumes, check migrations/init.sql is saved, then run docker compose up -d again.

Help me debug my Docker containers

Your infrastructure is live. Next up, you will build the Order API that writes payment events to the database using the transactional outbox pattern.

Build the Order API with Transactional Outbox

Your Docker stack is running and the database schema is in place. Now it's time to build the first service in the pipeline: the Order API.

The most common reliability bug in event-driven systems is the dual-write problem. Your service writes to the database AND publishes a message to Kafka. If the database write succeeds but the Kafka publish fails, your data is inconsistent forever. The transactional outbox pattern solves this by writing both the business record and the event into the same database transaction. A separate process (the poller, which you'll build in the next step) handles publishing.

In this step, get ready to:
  • Create shared database pool and Kafka client modules.
  • Build an Express API with a transactional outbox write.
  • Test the endpoint with curl and verify the outbox row.
Create the shared database pool

All three services in this pipeline need to connect to the same PostgreSQL database. Instead of duplicating connection logic, you'll create a shared module that exports a connection pool.

  • Create the shared directory and the database pool file by running these commands:
mkdir -p src/shared
mkdir -p src/order-api
  • Create a new file called src/shared/db.ts and add the following code:
import pg from 'pg';

// Create a connection pool to reuse across requests
const pool = new pg.Pool({
  host: 'localhost',
  port: 5432,
  user: 'pipeline',
  password: 'pipeline',
  database: 'payments',
});

export default pool;

What does this code do?

A connection pool keeps a set of database connections open and ready to use. Instead of opening a new connection for every request (slow and resource-heavy), the pool hands out existing connections and recycles them when done.

The credentials match the POSTGRES_USER, POSTGRES_PASSWORD, and POSTGRES_DB you set in docker-compose.yml.

  • Save src/shared/db.ts.
Create the shared Kafka client

Multiple services also need Kafka access. You'll create one shared module that exports a configured producer and consumer.

  • Create a new file called src/shared/kafka.ts and add the following code:
import { Kafka } from 'kafkajs';

// Single Kafka instance shared across services
const kafka = new Kafka({
  clientId: 'payment-pipeline',
  brokers: ['localhost:9092'],
});

// Pre-configured producer and consumer for the pipeline
export const producer = kafka.producer();
export const consumer = kafka.consumer({ groupId: 'payment-consumer-group' });
export default kafka;

What does this code do?

The Kafka constructor from KafkaJS connects to the Redpanda broker running on port 9092. Redpanda is Kafka API-compatible, so KafkaJS treats it exactly like a Kafka cluster.

The groupId on the consumer determines which consumer group receives messages. All consumers with the same group ID share the workload across topic partitions.

  • Save src/shared/kafka.ts.
Create the OpenTelemetry tracing module

The Order API imports a tracing module at startup to enable distributed tracing across all services. You'll create this shared module now so the API can start cleanly.

  • Create a new file called src/shared/tracing.ts and add the following code:
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
import { PgInstrumentation } from '@opentelemetry/instrumentation-pg';
import { resourceFromAttributes } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';

export function initTracing(serviceName: string) {
  // Configure the OpenTelemetry SDK with OTLP exporter pointing to Jaeger
  const sdk = new NodeSDK({
    resource: resourceFromAttributes({
      [ATTR_SERVICE_NAME]: serviceName,
    }),
    traceExporter: new OTLPTraceExporter({
      url: 'http://localhost:4318/v1/traces',
    }),
    instrumentations: [
      new HttpInstrumentation(),
      new PgInstrumentation(),
    ],
  });

  sdk.start();
  return sdk;
}

What does this code do?

The initTracing function initializes OpenTelemetry with automatic instrumentation for HTTP requests and PostgreSQL queries. Every database call and HTTP request is automatically captured as a trace span.

The OTLPTraceExporter sends trace data to Jaeger on port 4318. You'll explore these traces in a later step. For now, the module is set up so all services report telemetry from the start.

  • Save src/shared/tracing.ts.
  • Confirm your files exist by running:
ls src/shared/

You should see db.ts, kafka.ts, and tracing.ts listed.

Missing one of the files?

Make sure you created them inside src/shared/ and not in the project root. Check that the src/shared directory exists with ls src/. If it doesn't, re-run the mkdir -p src/shared command.

Help me debug my shared module setup.

✔️ Awesome, I've got everything!

Great. Make sure all three files are saved before moving on.

ⓧ I'd like to double check the full code

import pg from 'pg';

// Create a connection pool to reuse across requests
const pool = new pg.Pool({
  host: 'localhost',
  port: 5432,
  user: 'pipeline',
  password: 'pipeline',
  database: 'payments',
});

export default pool;
import { Kafka } from 'kafkajs';

// Single Kafka instance shared across services
const kafka = new Kafka({
  clientId: 'payment-pipeline',
  brokers: ['localhost:9092'],
});

// Pre-configured producer and consumer for the pipeline
export const producer = kafka.producer();
export const consumer = kafka.consumer({ groupId: 'payment-consumer-group' });
export default kafka;
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
import { PgInstrumentation } from '@opentelemetry/instrumentation-pg';
import { resourceFromAttributes } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';

export function initTracing(serviceName: string) {
  // Configure the OpenTelemetry SDK with OTLP exporter pointing to Jaeger
  const sdk = new NodeSDK({
    resource: resourceFromAttributes({
      [ATTR_SERVICE_NAME]: serviceName,
    }),
    traceExporter: new OTLPTraceExporter({
      url: 'http://localhost:4318/v1/traces',
    }),
    instrumentations: [
      new HttpInstrumentation(),
      new PgInstrumentation(),
    ],
  });

  sdk.start();
  return sdk;
}
Build the Express API with the POST /orders endpoint

This is the core of the transactional outbox pattern. The endpoint writes both the order record and the outbox event in a single database transaction. If either insert fails, both roll back.

  • Create a new file called src/order-api/index.ts and add the following code:
import { initTracing } from '../shared/tracing.js';
initTracing('order-api');

import express from 'express';
import { randomUUID } from 'crypto';
import pool from '../shared/db.js';

const app = express();
app.use(express.json());

app.post('/orders', async (req, res) => {
  const { customerId, amount, currency } = req.body;
  const orderId = randomUUID();
  const eventId = randomUUID();

  // Acquire a dedicated client for the transaction
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    // Insert the order record
    await client.query(
      `INSERT INTO orders (id, customer_id, amount, currency, status)
       VALUES ($1, $2, $3, $4, 'pending')`,
      [orderId, customerId, amount, currency || 'USD']
    );

    // Insert the outbox event in the SAME transaction
    await client.query(
      `INSERT INTO outbox (id, aggregate_id, event_type, payload, status)
       VALUES ($1, $2, 'order.created', $3, 'pending')`,
      [eventId, orderId, JSON.stringify({ orderId, customerId, amount, currency: currency || 'USD' })]
    );

    await client.query('COMMIT');
    res.status(201).json({ orderId, eventId });
  } catch (error) {
    await client.query('ROLLBACK');
    console.error('Failed to create order:', error);
    res.status(500).json({ error: 'Internal server error' });
  } finally {
    client.release();
  }
});

app.listen(3001, () => {
  console.log('Order API running on http://localhost:3001');
});

What does this code do?

The initTracing('order-api') call must come before all other imports. It registers instrumentation hooks that only work if loaded before the libraries they patch.

Inside the POST /orders handler, pool.connect() acquires a dedicated client from the pool. Using a dedicated client (instead of pool.query()) is required for transactions because BEGIN, COMMIT, and ROLLBACK must all run on the same connection.

The two INSERT statements write to different tables but share the same transaction. If the outbox insert fails, the order insert rolls back too. This guarantees the event is always recorded alongside the business data.

Why not publish directly to Kafka?

If you insert the order and then publish to Kafka as two separate steps, the publish can fail after the database commit. Your order exists but no event fires. Downstream services never learn about it.

The outbox pattern avoids this by keeping everything in one atomic database write. A separate poller reads the outbox and publishes to Kafka. If publishing fails, the poller retries. The event is never lost because it lives in the database until confirmed published.

  • Save src/order-api/index.ts.

Seeing red squiggly lines in VS Code?

TypeScript may show import errors for .js extensions. This is expected. The moduleResolution: "bundler" setting in tsconfig.json combined with tsx resolves .js imports to their .ts source files at runtime. The code will run correctly.

Help me fix TypeScript import errors in my order API.

✔️ Awesome, I've got everything!

Great. Double check that src/order-api/index.ts is saved before testing.

ⓧ I'd like to double check the full code

import { initTracing } from '../shared/tracing.js';
initTracing('order-api');

import express from 'express';
import { randomUUID } from 'crypto';
import pool from '../shared/db.js';

const app = express();
app.use(express.json());

app.post('/orders', async (req, res) => {
  const { customerId, amount, currency } = req.body;
  const orderId = randomUUID();
  const eventId = randomUUID();

  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    await client.query(
      `INSERT INTO orders (id, customer_id, amount, currency, status)
       VALUES ($1, $2, $3, $4, 'pending')`,
      [orderId, customerId, amount, currency || 'USD']
    );

    await client.query(
      `INSERT INTO outbox (id, aggregate_id, event_type, payload, status)
       VALUES ($1, $2, 'order.created', $3, 'pending')`,
      [eventId, orderId, JSON.stringify({ orderId, customerId, amount, currency: currency || 'USD' })]
    );

    await client.query('COMMIT');
    res.status(201).json({ orderId, eventId });
  } catch (error) {
    await client.query('ROLLBACK');
    console.error('Failed to create order:', error);
    res.status(500).json({ error: 'Internal server error' });
  } finally {
    client.release();
  }
});

app.listen(3001, () => {
  console.log('Order API running on http://localhost:3001');
});
Test the endpoint

Time to prove the transactional outbox works. You'll start the API, send a test order, and verify that both the order and the outbox event were written atomically.

  • Start the Order API by running:
npm run order-api

You should see Order API running on http://localhost:3001 in the terminal.

  • Open a second terminal. Send a test order by running:
curl -X POST http://localhost:3001/orders \
  -H "Content-Type: application/json" \
  -d '{"customerId": "cust-001", "amount": 99.99, "currency": "USD"}'

You should see a JSON response with orderId and eventId fields. The HTTP status is 201 (Created).

  • Verify the outbox row was written by querying the database:
docker compose exec postgres psql -U pipeline -d payments -c "SELECT id, aggregate_id, event_type, status FROM outbox;"

You should see one row with event_type of order.created and status of pending. This proves the outbox event was written in the same transaction as the order.

Getting connection refused on curl?

Make sure the first terminal is still showing Order API running on http://localhost:3001. If the process crashed, check for typos in src/order-api/index.ts or src/shared/db.ts. The most common issue is a missing comma in the pool configuration object.

If you see a database connection error, verify your Docker containers are running with docker compose ps. The postgres service must show a healthy status.

Help me debug my Order API startup error.

Your Order API is writing events atomically. Next up, you'll build the Outbox Poller that reads these pending events and publishes them to Kafka.

Build the Outbox Poller

Your Order API is writing events to the outbox table inside the same database transaction as the order itself. That guarantees the event will never be lost, even if Kafka is temporarily unreachable.

Now you need a separate process that watches the outbox table for pending events and publishes them to Kafka. This is the second half of the transactional outbox pattern: a poller that reliably drains the outbox into the message broker, retrying on failure.

In this step, get ready to:
  • Create a polling loop that queries pending outbox events every second.
  • Publish events to the Kafka topic with retry protection.
  • Verify the full pipeline from order creation to message delivery.
Create the poller loop

The poller is a standalone service that runs on a timer. Every cycle, it queries the outbox for pending events, publishes each one to Kafka, and marks it as published.

  • Create a new file at src/outbox-poller/index.ts and add the following code:
import { initTracing } from '../shared/tracing.js';
initTracing('outbox-poller');

import { context, propagation, trace } from '@opentelemetry/api';
import pool from '../shared/db.js';
import { producer } from '../shared/kafka.js';

const POLL_INTERVAL_MS = 1000;
const MAX_RETRIES = 5;

async function pollOutbox() {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    const result = await client.query(
      `SELECT id, aggregate_id, event_type, payload
       FROM outbox
       WHERE status = 'pending' AND retries < $1
       ORDER BY created_at ASC
       LIMIT 10
       FOR UPDATE SKIP LOCKED`,
      [MAX_RETRIES]
    );

    for (const row of result.rows) {
      const tracer = trace.getTracer('outbox-poller');
      const span = tracer.startSpan('publish-outbox-event');

      try {
        const headers: Record<string, string> = {};
        propagation.inject(context.active(), headers);

        await producer.send({
          topic: 'orders.events',
          messages: [{
            key: row.aggregate_id,
            value: JSON.stringify({
              eventId: row.id,
              eventType: row.event_type,
              payload: row.payload,
              timestamp: new Date().toISOString(),
            }),
            headers,
          }],
        });

        await client.query(
          `UPDATE outbox SET status = 'published', published_at = NOW() WHERE id = $1`,
          [row.id]
        );

        span.end();
      } catch (error) {
        await client.query(
          `UPDATE outbox SET retries = retries + 1 WHERE id = $1`,
          [row.id]
        );
        span.recordException(error as Error);
        span.end();
        console.error(`Failed to publish event ${row.id}:`, error);
      }
    }

    await client.query('COMMIT');
  } catch (error) {
    await client.query('ROLLBACK');
    console.error('Poller error:', error);
  } finally {
    client.release();
  }
}

async function start() {
  await producer.connect();
  console.log('Outbox Poller started, polling every', POLL_INTERVAL_MS, 'ms');
  setInterval(pollOutbox, POLL_INTERVAL_MS);
}

start().catch(console.error);

What does this code do?

The poller queries up to 10 pending events per cycle using SELECT ... FOR UPDATE SKIP LOCKED. This PostgreSQL feature locks the selected rows so that if you ever run multiple poller instances, they won't process the same event twice.

For each event, the poller creates an OpenTelemetry span and injects the trace context into the Kafka message headers. This allows traces to flow across the async Kafka boundary, connecting the poller's work to the downstream consumer.

On success, the row is marked 'published' with a timestamp. On failure, the retries counter increments and the event stays pending for the next poll cycle.

  • Save src/outbox-poller/index.ts.

File not saving?

Make sure the src/outbox-poller/ directory exists. Create it first if your editor did not auto-create it on save.

help me create the outbox-poller directory structure

Handle failures and retries

The poller is designed to be resilient. If Kafka is temporarily down or a network blip interrupts publishing, the event simply stays in 'pending' state.

How does the retry mechanism work?

Each time a publish attempt fails, the poller increments the event's retries counter in the database. The next poll cycle picks it up again because it still has status = 'pending'.

The query condition retries < 5 (the MAX_RETRIES constant) acts as a circuit breaker for individual events. After 5 failed attempts, the event is abandoned in place. In production, you would alert on these stale events and investigate the root cause.

This approach gives you at-least-once delivery without requiring distributed transactions between PostgreSQL and Kafka.

Why polling instead of database triggers?

A poll-based approach is simple to reason about, easy to scale horizontally, and does not require any database extensions or CDC tooling. In production systems, you might replace this with a change data capture (CDC) tool like Debezium for lower latency. For this project, polling every second is fast enough and keeps the architecture straightforward.

Verify end-to-end flow

Time to prove the full pipeline works: create an order, watch the poller pick it up, and confirm the message lands in Redpanda.

  • Make sure your Docker Compose stack is still running.
  • Open a terminal and start the Order API by running:
npm run order-api
  • Open a second terminal and start the Outbox Poller by running:
npm run outbox-poller

You should see Outbox Poller started, polling every 1000 ms in the terminal.

  • Open a third terminal and create an order by running:
curl -X POST http://localhost:3001/orders -H "Content-Type: application/json" -d "{\"customerId\": \"cust-123\", \"amount\": 49.99, \"currency\": \"USD\"}"

You should see a JSON response with orderId and eventId fields. Within one second, the poller terminal should log that it published the event.

  • Verify the message arrived in the Kafka topic by running:
docker compose exec redpanda rpk topic consume orders.events --num 1

What does this command do?

This runs rpk (the Redpanda CLI bundled inside the container) to read messages from the orders.events topic. The --num 1 flag tells it to print one message and exit.

You should see a JSON object containing your order's eventId, eventType, payload, and timestamp fields printed to the terminal.

Not seeing the message?

Check that the poller terminal shows no errors. If you see connection refused errors, verify Redpanda is healthy with docker compose ps.

If the poller started before the Order API had created an order, the outbox table might be empty. Resend the curl request and wait one second for the poller to pick it up.

why isn't my outbox poller publishing events to Kafka?

  • Press Ctrl+C in the Order API and Outbox Poller terminals to stop both services.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/outbox-poller/index.ts and that your rpk output confirmed the message arrived.

ⓧ I'd like to double check the full code

Here is the complete src/outbox-poller/index.ts file:

import { initTracing } from '../shared/tracing.js';
initTracing('outbox-poller');

import { context, propagation, trace } from '@opentelemetry/api';
import pool from '../shared/db.js';
import { producer } from '../shared/kafka.js';

const POLL_INTERVAL_MS = 1000;
const MAX_RETRIES = 5;

async function pollOutbox() {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    const result = await client.query(
      `SELECT id, aggregate_id, event_type, payload
       FROM outbox
       WHERE status = 'pending' AND retries < $1
       ORDER BY created_at ASC
       LIMIT 10
       FOR UPDATE SKIP LOCKED`,
      [MAX_RETRIES]
    );

    for (const row of result.rows) {
      const tracer = trace.getTracer('outbox-poller');
      const span = tracer.startSpan('publish-outbox-event');

      try {
        const headers: Record<string, string> = {};
        propagation.inject(context.active(), headers);

        await producer.send({
          topic: 'orders.events',
          messages: [{
            key: row.aggregate_id,
            value: JSON.stringify({
              eventId: row.id,
              eventType: row.event_type,
              payload: row.payload,
              timestamp: new Date().toISOString(),
            }),
            headers,
          }],
        });

        await client.query(
          `UPDATE outbox SET status = 'published', published_at = NOW() WHERE id = $1`,
          [row.id]
        );

        span.end();
      } catch (error) {
        await client.query(
          `UPDATE outbox SET retries = retries + 1 WHERE id = $1`,
          [row.id]
        );
        span.recordException(error as Error);
        span.end();
        console.error(`Failed to publish event ${row.id}:`, error);
      }
    }

    await client.query('COMMIT');
  } catch (error) {
    await client.query('ROLLBACK');
    console.error('Poller error:', error);
  } finally {
    client.release();
  }
}

async function start() {
  await producer.connect();
  console.log('Outbox Poller started, polling every', POLL_INTERVAL_MS, 'ms');
  setInterval(pollOutbox, POLL_INTERVAL_MS);
}

start().catch(console.error);

Your outbox poller is draining events from the database and delivering them to Kafka reliably. Next up, you will build the Payment Consumer that processes these events exactly once using idempotency keys and circuit breaker protection.

Build the Idempotent Payment Consumer with Circuit Breaker

Your Outbox Poller is reliably publishing events to Kafka. But what happens on the receiving end when the consumer processes the same message twice, or when a downstream payment gateway goes down?

In this step, you will build a Payment Consumer that solves both problems. It checks a deduplication table before processing each event, guaranteeing exactly-once semantics. It also wraps payment calls in a circuit breaker that trips open under sustained failure, routing poisoned messages to a dead letter queue instead of crashing the entire consumer.

In this step, get ready to:
  • Create a Kafka consumer with idempotency checking via the processed_events table.
  • Wrap payment processing in a circuit breaker with dead letter queue fallback.
  • Test fault tolerance by observing circuit breaker state transitions and DLQ routing.
Create the Kafka consumer with the payment processing function

The consumer needs a function that actually processes payments. This function simulates a real-world payment gateway that sometimes fails.

  • Create a new file at src/payment-consumer/index.ts by adding the following code:
import { initTracing } from '../shared/tracing.js';
initTracing('payment-consumer');

import { context, propagation, trace } from '@opentelemetry/api';
import CircuitBreaker from 'opossum';
import pool from '../shared/db.js';
import { consumer, producer } from '../shared/kafka.js';

const DLQ_TOPIC = 'orders.events.dlq';

async function processPayment(payload: { orderId: string; customerId: string; amount: number; currency: string }) {
  // Simulate payment processing with occasional failures
  if (Math.random() < 0.3) {
    throw new Error('Payment gateway timeout');
  }

  await pool.query(
    `UPDATE orders SET status = 'paid' WHERE id = $1`,
    [payload.orderId]
  );

  console.log(`Payment processed for order ${payload.orderId}: ${payload.amount} ${payload.currency}`);
  return { success: true, orderId: payload.orderId };
}

What does this code do?

The tracing initialization must come before all other imports so that OpenTelemetry can patch modules before they load.

The processPayment function simulates a real payment gateway. The Math.random() < 0.3 check means roughly 30% of calls will throw an error. On success, it updates the order status to paid in PostgreSQL.

The DLQ_TOPIC constant defines where failed messages go when the circuit breaker trips. DLQ stands for Dead Letter Queue.

Wrap the payment processing in a circuit breaker

A circuit breaker monitors the success rate of a function. When failures exceed a threshold, it "opens" the circuit and stops calling the function entirely. This prevents a failing downstream service from overwhelming your consumer.

  • Add the circuit breaker configuration below the processPayment function in src/payment-consumer/index.ts:
const breaker = new CircuitBreaker(processPayment, {
  timeout: 5000,
  errorThresholdPercentage: 50,
  resetTimeout: 10000,
});

breaker.on('open', () => console.log('CIRCUIT BREAKER: OPEN - failing fast'));
breaker.on('halfOpen', () => console.log('CIRCUIT BREAKER: HALF-OPEN - testing'));
breaker.on('close', () => console.log('CIRCUIT BREAKER: CLOSED - healthy'));

breaker.fallback(async (payload) => {
  await producer.send({
    topic: DLQ_TOPIC,
    messages: [{ key: payload.orderId, value: JSON.stringify(payload) }],
  });
  console.log(`Routed to DLQ: order ${payload.orderId}`);
  return { success: false, dlq: true };
});

What does the circuit breaker do?

  • timeout: 5000 means if processPayment takes longer than 5 seconds, it counts as a failure.
  • errorThresholdPercentage: 50 means once 50% of recent calls fail, the circuit opens.
  • resetTimeout: 10000 means after 10 seconds in the open state, the breaker enters half-open to test one request.
  • The fallback function runs whenever the breaker is open or the action fails. It publishes the failed payload to the DLQ topic so nothing is lost.

The three state-change listeners log transitions so you can observe the breaker cycling through CLOSED, OPEN, and HALF-OPEN in real time.

Why a circuit breaker instead of simple retries?

Simple retries hammer a failing service with repeated requests, making the outage worse. A circuit breaker detects the failure pattern and stops calling entirely, giving the downstream service time to recover.

When the resetTimeout expires, the breaker lets a single test request through. If it succeeds, the circuit closes and normal traffic resumes. If it fails, the circuit stays open for another cycle.

  • Now add the start() function below the fallback configuration. This is the consumer loop with idempotency checking:
async function start() {
  await producer.connect();
  await consumer.connect();
  await consumer.subscribe({ topic: 'orders.events', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ message }) => {
      if (!message.value) return;

      // Extract trace context from headers
      const carrier: Record<string, string> = {};
      if (message.headers) {
        for (const [key, value] of Object.entries(message.headers)) {
          if (value) carrier[key] = value.toString();
        }
      }
      const extractedContext = propagation.extract(context.active(), carrier);

      const tracer = trace.getTracer('payment-consumer');
      const span = tracer.startSpan('process-payment', undefined, extractedContext);

      try {
        const event = JSON.parse(message.value.toString());
        const eventId = event.eventId;

        // Idempotency check
        const existing = await pool.query(
          `SELECT event_id FROM processed_events WHERE event_id = $1`,
          [eventId]
        );

        if (existing.rows.length > 0) {
          console.log(`Skipping duplicate event: ${eventId}`);
          span.setAttribute('duplicate', true);
          span.end();
          return;
        }

        // Process with circuit breaker
        await breaker.fire(event.payload);

        // Record as processed
        await pool.query(
          `INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING`,
          [eventId]
        );

        span.end();
      } catch (error) {
        span.recordException(error as Error);
        span.end();
        console.error('Consumer error:', error);
      }
    },
  });

  console.log('Payment Consumer started');
}

start().catch(console.error);

How does the idempotency check work?

Before processing any event, the consumer queries the processed_events table for the event ID. If a row exists, the event was already handled and the consumer skips it. This prevents double-processing when Kafka redelivers a message.

After successful processing, the event ID is recorded with INSERT ... ON CONFLICT DO NOTHING. The ON CONFLICT clause handles rare race conditions where two consumer instances try to insert the same ID simultaneously.

Trace context is extracted from Kafka message headers so that Jaeger can connect this consumer span back to the original Order API request. The span is named process-payment for easy filtering in the Jaeger UI.

  • Save src/payment-consumer/index.ts.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

import { initTracing } from '../shared/tracing.js';
initTracing('payment-consumer');

import { context, propagation, trace } from '@opentelemetry/api';
import CircuitBreaker from 'opossum';
import pool from '../shared/db.js';
import { consumer, producer } from '../shared/kafka.js';

const DLQ_TOPIC = 'orders.events.dlq';

async function processPayment(payload: { orderId: string; customerId: string; amount: number; currency: string }) {
  // Simulate payment processing with occasional failures
  if (Math.random() < 0.3) {
    throw new Error('Payment gateway timeout');
  }

  await pool.query(
    `UPDATE orders SET status = 'paid' WHERE id = $1`,
    [payload.orderId]
  );

  console.log(`Payment processed for order ${payload.orderId}: ${payload.amount} ${payload.currency}`);
  return { success: true, orderId: payload.orderId };
}

const breaker = new CircuitBreaker(processPayment, {
  timeout: 5000,
  errorThresholdPercentage: 50,
  resetTimeout: 10000,
});

breaker.on('open', () => console.log('CIRCUIT BREAKER: OPEN - failing fast'));
breaker.on('halfOpen', () => console.log('CIRCUIT BREAKER: HALF-OPEN - testing'));
breaker.on('close', () => console.log('CIRCUIT BREAKER: CLOSED - healthy'));

breaker.fallback(async (payload) => {
  await producer.send({
    topic: DLQ_TOPIC,
    messages: [{ key: payload.orderId, value: JSON.stringify(payload) }],
  });
  console.log(`Routed to DLQ: order ${payload.orderId}`);
  return { success: false, dlq: true };
});

async function start() {
  await producer.connect();
  await consumer.connect();
  await consumer.subscribe({ topic: 'orders.events', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ message }) => {
      if (!message.value) return;

      // Extract trace context from headers
      const carrier: Record<string, string> = {};
      if (message.headers) {
        for (const [key, value] of Object.entries(message.headers)) {
          if (value) carrier[key] = value.toString();
        }
      }
      const extractedContext = propagation.extract(context.active(), carrier);

      const tracer = trace.getTracer('payment-consumer');
      const span = tracer.startSpan('process-payment', undefined, extractedContext);

      try {
        const event = JSON.parse(message.value.toString());
        const eventId = event.eventId;

        // Idempotency check
        const existing = await pool.query(
          `SELECT event_id FROM processed_events WHERE event_id = $1`,
          [eventId]
        );

        if (existing.rows.length > 0) {
          console.log(`Skipping duplicate event: ${eventId}`);
          span.setAttribute('duplicate', true);
          span.end();
          return;
        }

        // Process with circuit breaker
        await breaker.fire(event.payload);

        // Record as processed
        await pool.query(
          `INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING`,
          [eventId]
        );

        span.end();
      } catch (error) {
        span.recordException(error as Error);
        span.end();
        console.error('Consumer error:', error);
      }
    },
  });

  console.log('Payment Consumer started');
}

start().catch(console.error);

Seeing a TypeScript error about opossum types?

Make sure @types/opossum is installed as a dev dependency. Check your package.json for "@types/opossum": "8.1.7" in devDependencies. If it is missing, run npm install again from your project root.

help me fix opossum TypeScript type errors in my payment consumer

Test fault tolerance

With all three services ready, you can now observe the full pipeline including circuit breaker behavior under simulated failure.

  • Make sure your Docker Compose stack is still running. If not, start it by running:
docker compose up -d
  • Open a new terminal and start the Order API by running:
npm run order-api
  • Open another terminal and start the Outbox Poller by running:
npm run outbox-poller
  • Open a third terminal and start the Payment Consumer by running:
npm run payment-consumer
  • Open a fourth terminal and send several orders rapidly to trigger the circuit breaker. Run this command multiple times:
curl -X POST http://localhost:3001/orders -H "Content-Type: application/json" -d '{"customerId": "cust-1", "amount": 99.99, "currency": "USD"}'
  • Run the curl command 10-15 times in quick succession.
  • Watch the Payment Consumer terminal for circuit breaker state transitions.

You should see the consumer log messages like Payment processed for order... for successful payments. As failures accumulate (around 30% of the time), you will eventually see CIRCUIT BREAKER: OPEN - failing fast followed by Routed to DLQ messages.

After 10 seconds (the resetTimeout), you will see CIRCUIT BREAKER: HALF-OPEN - testing as the breaker tries one request. If it succeeds, CIRCUIT BREAKER: CLOSED - healthy appears and normal processing resumes.

  • Verify messages landed in the DLQ by running:
docker exec -it $(docker ps -qf "ancestor=redpandadata/redpanda:latest") rpk topic consume orders.events.dlq --num 1

What should I see?

The DLQ topic should contain JSON payloads of orders that failed processing. Each message includes the orderId, customerId, amount, and currency fields. These are the messages the circuit breaker routed to the DLQ instead of dropping.

If you do not see DLQ messages, send more orders. The 30% simulated failure rate means you need enough volume for the errorThresholdPercentage of 50% to trigger.

Circuit breaker not opening?

The breaker needs a rolling window of requests before it calculates the error percentage. If you send orders slowly (one every few seconds), the window may not accumulate enough failures. Send 10+ orders in rapid succession to trigger the threshold.

Also confirm your consumer is actually running and subscribing to the correct topic. Check for Payment Consumer started in the terminal output.

help me debug why my circuit breaker is not opening after sending multiple orders

Your pipeline now handles the two most dangerous failure modes in event-driven systems: duplicate processing and cascading downstream failures. Next up, you will add distributed tracing so you can visualize the entire flow from order creation through payment processing in Jaeger.

Add Distributed Tracing with OpenTelemetry

Your pipeline processes payments reliably, but when something goes wrong across service boundaries, you have no way to follow a single request from start to finish.

In this step, you will create the OpenTelemetry instrumentation that connects spans across all three services. By the end, you will see a single trace in Jaeger that shows the full lifecycle of a payment event as it flows through Kafka.

In this step, get ready to:
  • Create the shared OpenTelemetry instrumentation file that all three services depend on.
  • Understand how trace context propagates across Kafka message boundaries.
  • Verify end-to-end distributed traces in the Jaeger UI.
Create the OpenTelemetry instrumentation file

All three services already import a function called initTracing from ../shared/tracing.js. That file does not exist yet. Creating it is the single missing piece that lights up distributed tracing across your entire pipeline.

  • Create a new file called tracing.ts inside the src/shared/ folder by running:
touch src/shared/tracing.ts
  • Confirm it was created by running:
ls src/shared/

You should see db.ts, kafka.ts, and tracing.ts listed.

  • Open src/shared/tracing.ts in your editor and paste in the following code:
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
import { PgInstrumentation } from '@opentelemetry/instrumentation-pg';
import { resourceFromAttributes } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';

export function initTracing(serviceName: string) {
  const sdk = new NodeSDK({
    resource: resourceFromAttributes({
      [ATTR_SERVICE_NAME]: serviceName,
    }),
    traceExporter: new OTLPTraceExporter({
      url: 'http://localhost:4318/v1/traces',
    }),
    instrumentations: [
      new HttpInstrumentation(),
      new PgInstrumentation(),
    ],
  });

  sdk.start();
  return sdk;
}

What does this code do?

This file configures the OpenTelemetry SDK for each service that calls it.

  • resourceFromAttributes tags every span with the service name so Jaeger can group spans by origin.
  • OTLPTraceExporter sends trace data to Jaeger's OTLP collector endpoint at http://localhost:4318/v1/traces.
  • HttpInstrumentation automatically creates spans for incoming and outgoing HTTP requests.
  • PgInstrumentation automatically creates spans for every PostgreSQL query.
  • Save src/shared/tracing.ts.

Why must initTracing be the first import?

OpenTelemetry works by monkey-patching libraries (like pg and http) before they are imported. If you import express or pg first, OpenTelemetry misses the patching window and those libraries produce no spans.

That is why every service file calls initTracing() on line 2, before any other imports.

Seeing TypeScript import errors?

Make sure all the OpenTelemetry packages are installed. They were added to package.json in Step 1. If you see unresolved module errors, run npm install again from your project root.

Help me debug my tracing.ts import errors

✔️ Awesome, I've got everything!

Great. Double check you have saved src/shared/tracing.ts before moving on.

ⓧ I'd like to double check the full code

Here is the complete src/shared/tracing.ts file:

import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
import { PgInstrumentation } from '@opentelemetry/instrumentation-pg';
import { resourceFromAttributes } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';

export function initTracing(serviceName: string) {
  const sdk = new NodeSDK({
    resource: resourceFromAttributes({
      [ATTR_SERVICE_NAME]: serviceName,
    }),
    traceExporter: new OTLPTraceExporter({
      url: 'http://localhost:4318/v1/traces',
    }),
    instrumentations: [
      new HttpInstrumentation(),
      new PgInstrumentation(),
    ],
  });

  sdk.start();
  return sdk;
}
Understand trace context propagation across Kafka

Automatic instrumentation handles HTTP and database spans. But Kafka is an asynchronous boundary where traces would normally break apart. Your services already contain manual propagation code to bridge this gap.

Open src/outbox-poller/index.ts and find this section inside the for loop:

const headers: Record<string, string> = {};
propagation.inject(context.active(), headers);

await producer.send({
  topic: 'orders.events',
  messages: [{
    key: row.aggregate_id,
    value: JSON.stringify({
      eventId: row.id,
      eventType: row.event_type,
      payload: row.payload,
      timestamp: new Date().toISOString(),
    }),
    headers,
  }],
});

What is propagation.inject doing?

propagation.inject serializes the current trace ID and span ID into the headers object. When the Kafka message is sent, those headers travel alongside the payload.

This is the producer side of context propagation. Without it, the consumer would start a brand-new trace with no connection to the original request.

Now open src/payment-consumer/index.ts and find this section inside eachMessage:

// Extract trace context from headers
const carrier: Record<string, string> = {};
if (message.headers) {
  for (const [key, value] of Object.entries(message.headers)) {
    if (value) carrier[key] = value.toString();
  }
}
const extractedContext = propagation.extract(context.active(), carrier);

const tracer = trace.getTracer('payment-consumer');
const span = tracer.startSpan('process-payment', undefined, extractedContext);

What is propagation.extract doing?

propagation.extract reads the trace context from the Kafka message headers and reconstructs it. The new span is started as a child of the extracted context, which means Jaeger links it to the same trace that started back in the Order API.

Together, inject on the producer and extract on the consumer create a continuous trace across the asynchronous Kafka boundary.

Why not use automatic Kafka instrumentation?

OpenTelemetry does not ship a stable automatic instrumentation for KafkaJS. Manual injection and extraction gives you full control over which spans are created and how they connect across the message boundary. It also makes the propagation explicit and easier to debug.

Verify traces in Jaeger

With tracing.ts in place, all three services will now emit spans to Jaeger. Time to run the full pipeline and confirm distributed traces appear.

  • Make sure your Docker Compose stack is still running. If it is not, start it by running:
docker compose up -d
  • Open three separate terminal windows. Start each service in its own terminal by running these commands (one per terminal):
npm run order-api
npm run outbox-poller
npm run payment-consumer

You should see each service log its startup message.

  • Send a test order to trigger the full pipeline by running this in a fourth terminal:
curl -X POST http://localhost:3001/orders -H "Content-Type: application/json" -d "{\"customerId\": \"cust-123\", \"amount\": 49.99, \"currency\": \"USD\"}"

You should see a 201 response with an orderId and eventId. Wait 2-3 seconds for the poller to pick up the event and the consumer to process it.

  • Open the Jaeger UI in your browser at http://localhost:16686.
  • In the Search page, select order-api from the Service dropdown.
  • Click Find Traces.
  • Click on one of the traces in the results list to open it.

You should see a trace with spans from order-api, outbox-poller, and payment-consumer. The spans are connected through the Kafka message boundary, showing the full lifecycle of your payment event.

What should I see in the trace?

A healthy trace includes HTTP spans from the POST request, PostgreSQL query spans from the transactional writes, a publish-outbox-event span from the poller, and a process-payment span from the consumer. All connected under one trace ID.

Not seeing any services in Jaeger?

  • Confirm the Jaeger container is running with docker compose ps. Look for the jaeger service in healthy state.
  • Make sure you sent at least one request with curl after starting the services. Jaeger only shows services that have emitted traces.
  • Check that initTracing is the very first import in each service file. If other imports come before it, spans will not be captured.

Help me debug why my traces are not appearing in Jaeger

Your pipeline is now fully observable. Every payment event can be traced from the moment it enters the Order API, through the outbox and Kafka, all the way to the payment consumer. If something fails in production, you can pinpoint exactly where and why.

Secret mission

Add Prometheus Metrics and a Grafana Dashboard

Your pipeline is reliable and traceable. But can you prove it under pressure? Add real-time Prometheus metrics and a Grafana dashboard that visualizes circuit breaker state transitions, message throughput, and latency as they happen.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally via Docker, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker containers (Redpanda, PostgreSQL, Jaeger, Prometheus, Grafana).
  • Local project folder (payment-pipeline) with source code and node_modules.
  • Docker images pulled during the project.

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with the pipeline.

  • Your Docker containers continue running in the background.
  • The Jaeger UI stays accessible at http://localhost:16686.
  • Grafana stays accessible at http://localhost:3000.
  • Your project files remain on your Desktop.

Pause - I'll come back to this later

Stop the Docker containers to free up memory and CPU. Your project files and Docker images stay on disk so you can restart quickly.

  • Open your terminal and navigate to the payment-pipeline folder on your Desktop.
  • Stop all containers by running:
docker compose down

When you want to resume, run docker compose up -d from the same folder to restart everything.

Delete - I don't want to use this again

Remove all Docker containers, volumes, images, and the project folder entirely. This frees up all disk space used by this project.

  • Open your terminal and navigate to the payment-pipeline folder on your Desktop.
  • Stop and remove all containers and their volumes by running:
docker compose down -v
  • Remove the Docker images pulled for this project by running:
docker rmi redpandadata/redpanda:latest postgres:16 jaegertracing/all-in-one:latest prom/prometheus:latest grafana/grafana:latest
  • Navigate up to your Desktop and delete the project folder.

macOS

cd ~/Desktop
rm -rf payment-pipeline

Windows

cd ~/Desktop
rmdir /s /q payment-pipeline

Nice Work!

Nice Work!

You just built a production-grade, fault-tolerant payment processing pipeline from scratch. Three TypeScript services, a Kafka-compatible broker, and end-to-end observability prove you can architect distributed systems where data never gets lost and failures are isolated.

Here's what you accomplished:

  • Built a multi-service event-driven pipeline with guaranteed delivery via the transactional outbox pattern, eliminating dual-write failures between PostgreSQL and Kafka.
  • Implemented an idempotent consumer protected by a circuit breaker that degrades gracefully under failure and routes poisoned messages to a dead letter queue.
  • Instrumented the entire system with OpenTelemetry for end-to-end distributed tracing across async Kafka boundaries, visualized in Jaeger.
  • Secret Mission: Added Prometheus metrics and a Grafana dashboard showing real-time circuit breaker state transitions and message throughput.

Ready to quiz yourself?