Order Processing Pipeline with Redis

Build an event-driven order pipeline using Kafka, Redis, and Node.js.

Introduction

30 Second Summary

Every time you place an order online, dozens of systems spring into action behind the scenes. Your payment gets charged, inventory updates, a confirmation email fires off, and shipping gets notified. These systems don't wait for each other. They communicate through events, processing them reliably without losing a single order.

In this project, you will build an order processing pipeline using two Node.js processes that communicate through Apache Kafka, with Redis providing fast caching and duplicate detection.

What You'll Build

An event-driven order pipeline consisting of an Express.js API and a Kafka consumer worker, connected by Kafka for event streaming and Redis for speed.

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

  • An Express.js REST API with a Redis-cached product endpoint and an order endpoint that publishes events to Kafka.
  • A Kafka consumer worker that processes order events using a consumer group and prevents duplicate processing with Redis-based idempotency.
  • Local Docker containers running Redis and Kafka, giving you a production-like environment on your own machine.
  • Secret Mission: Add a dead-letter topic for failed orders and Redis-based rate limiting to handle real-world failure scenarios.

Are there any prerequisites?

You should have basic familiarity with Node.js and Express.js. You also need Docker Desktop installed and running on your machine.

Set Up Infrastructure

Before you can build an order processing pipeline, you need the two services it depends on: Redis for fast caching and Apache Kafka for durable event streaming.

In this step, you will spin up both services as Docker containers, then create a Node.js project with the packages you need to talk to them.

In this step, get ready to:
  • Start Redis and Kafka in Docker containers.
  • Initialize a Node.js project with the required packages.
  • Verify that your code can connect to both services.
Start Redis and Kafka containers

Docker lets you run Redis and Kafka locally without installing them on your machine directly. Each runs in its own isolated container.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Start a Redis container by running this command:
docker run -d --name redis -p 6379:6379 redis

What does this command do?

  • -d runs the container in the background (detached mode).
  • --name redis gives the container the name redis so you can reference it easily.
  • -p 6379:6379 maps port 6379 on your machine to port 6379 inside the container.
  • redis at the end is the official Redis Docker image.
  • Start a Kafka container by running this command:
docker run -d --name=kafka -p 9092:9092 apache/kafka

What does this command do?

This starts a single-node Kafka instance using KRaft mode (no Zookeeper needed). The apache/kafka image uses a default KRaft configuration automatically.

Port 9092 is Kafka's default broker port, mapped to your machine so your Node.js code can reach it.

  • Confirm both containers are running by running this command:
docker ps

You should see two rows in the output table, one with the name redis and one with the name kafka. Both should have a status starting with "Up".

Don't see both containers?

  • Make sure Docker Desktop is running. Open it from your system tray if it is not.
  • If you see a name conflict error, a container with that name already exists. Remove it first with docker rm -f redis or docker rm -f kafka, then re-run the command.

Still stuck? Help me troubleshoot my Docker containers not starting.

Initialize the Node.js project
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir order-pipeline
cd order-pipeline
  • Initialize a new Node.js project by running this command:
npm init -y
  • Install the three packages your project needs by running this command:
npm install express@5.2.1 redis@6.0.0 @confluentinc/kafka-javascript@1.9.1

What are these packages?

  • express@5.2.1 is the web framework for your API server.
  • redis@6.0.0 is the official Node.js client for Redis.
  • @confluentinc/kafka-javascript@1.9.1 is the Confluent JavaScript client for Apache Kafka. It includes pre-built native binaries so no C++ compiler is needed.

The install may take a minute because the Kafka package downloads a pre-built native binary for your platform.

Install taking a very long time or failing?

  • On macOS with an Intel chip, if the install hangs for more than 5 minutes, the package may be trying to compile from source. Try deleting node_modules and running the install command again.
  • Ensure you are using Node.js v20 or higher. Run node --version to check.

Need more help? Help me troubleshoot npm install for the Kafka package.

Now you need to configure package.json to use ES modules and add the npm scripts you will use throughout the project.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Select the order-pipeline folder on your Desktop, then click Open.
  • In the Explorer sidebar, click package.json to open it.
  • Replace the entire contents of package.json with the following:
{
  "name": "order-pipeline",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "api": "node src/api.js",
    "worker": "node src/worker.js",
    "test-connection": "node src/test-connection.js"
  },
  "dependencies": {
    "@confluentinc/kafka-javascript": "1.9.1",
    "express": "5.2.1",
    "redis": "6.0.0"
  }
}

What did you change?

  • "type": "module" enables ES module syntax (import/export) instead of CommonJS require().
  • The three scripts give you shortcuts to run your API, worker, and connection test without typing full paths.
  • Save package.json with Cmd+S (macOS) or Ctrl+S (Windows).
Verify connections

You will create three small files: a Redis client module, a Kafka client module, and a test script that uses both to confirm connectivity.

  • Create the src directory inside your order-pipeline folder. Open the integrated terminal in VS Code with Ctrl+` (backtick key, same on both platforms) and run:
mkdir src
  • Create the Redis client module. In the Explorer sidebar, right-click the src folder and select New File.
  • Name it redis-client.js and paste the following code:
import { createClient } from 'redis';

const redisClient = createClient({ url: 'redis://localhost:6379' });

redisClient.on('error', (err) => console.error('Redis Client Error:', err));

await redisClient.connect();

console.log('Connected to Redis');

export default redisClient;

What does this code do?

  • createClient initializes a Redis client pointing at your local container on port 6379.
  • The error event handler logs connection problems instead of crashing silently.
  • await redisClient.connect() establishes the connection. The redis package requires an explicit connect() call.
  • The client is exported so other files can reuse the same connection.
  • Save src/redis-client.js.
  • Create the Kafka client module. Right-click the src folder again and select New File.
  • Name it kafka-client.js and paste the following code:
import { KafkaJS } from '@confluentinc/kafka-javascript';

const kafka = new KafkaJS.Kafka({
  'bootstrap.servers': 'localhost:9092',
});

export default kafka;

What does this code do?

  • KafkaJS is the promisified API from the Confluent client. It gives you a familiar async/await interface.
  • bootstrap.servers tells the client where to find your local Kafka broker.
  • The kafka instance is exported so both the producer and consumer can share one configuration.
  • Save src/kafka-client.js.
  • Create the test connection script. Right-click the src folder and select New File.
  • Name it test-connection.js and paste the following code:
import redisClient from './redis-client.js';
import kafka from './kafka-client.js';

console.log('Testing Redis connection...');
await redisClient.set('test-key', 'hello');
const value = await redisClient.get('test-key');
console.log(`Redis test: ${value === 'hello' ? 'PASSED' : 'FAILED'}`);
await redisClient.del('test-key');

console.log('\nTesting Kafka connection...');
const admin = kafka.admin();
await admin.connect();
const topics = await admin.listTopics();
console.log(`Kafka test: PASSED (found ${topics.length} topics)`);
await admin.disconnect();

await redisClient.quit();
console.log('\nAll connections verified!');
process.exit(0);

What does this code do?

  • The Redis test writes a key, reads it back, and checks the value matches. Then it cleans up the test key.
  • The Kafka test creates an admin client, connects to the broker, and lists existing topics. If it connects without error, the test passes.
  • redisClient.quit() cleanly closes the Redis connection before the script exits.
  • Save src/test-connection.js.
  • Run the test script from your VS Code terminal:
npm run test-connection

You should see output ending with All connections verified! and both tests showing PASSED.

Seeing a connection refused error?

  • Run docker ps to verify both the redis and kafka containers show a status of "Up". If either is missing, restart it with docker start redis or docker start kafka.
  • If the Kafka test fails but Redis passes, Kafka may need a few seconds to fully start. Wait 10 seconds and run the test again.
  • Make sure your package.json has "type": "module". Without it, the import statements will fail.

Still stuck? Help me debug my Redis or Kafka connection test failing.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files in the src/ folder and that package.json matches the version above.

ⓧ I'd like to double check the full code

Here is each file as it should look at this point.

{
  "name": "order-pipeline",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "api": "node src/api.js",
    "worker": "node src/worker.js",
    "test-connection": "node src/test-connection.js"
  },
  "dependencies": {
    "@confluentinc/kafka-javascript": "1.9.1",
    "express": "5.2.1",
    "redis": "6.0.0"
  }
}
import { createClient } from 'redis';

const redisClient = createClient({ url: 'redis://localhost:6379' });

redisClient.on('error', (err) => console.error('Redis Client Error:', err));

await redisClient.connect();

console.log('Connected to Redis');

export default redisClient;
import { KafkaJS } from '@confluentinc/kafka-javascript';

const kafka = new KafkaJS.Kafka({
  'bootstrap.servers': 'localhost:9092',
});

export default kafka;
import redisClient from './redis-client.js';
import kafka from './kafka-client.js';

console.log('Testing Redis connection...');
await redisClient.set('test-key', 'hello');
const value = await redisClient.get('test-key');
console.log(`Redis test: ${value === 'hello' ? 'PASSED' : 'FAILED'}`);
await redisClient.del('test-key');

console.log('\nTesting Kafka connection...');
const admin = kafka.admin();
await admin.connect();
const topics = await admin.listTopics();
console.log(`Kafka test: PASSED (found ${topics.length} topics)`);
await admin.disconnect();

await redisClient.quit();
console.log('\nAll connections verified!');
process.exit(0);

Your infrastructure is ready. Redis and Kafka are running locally, and your project can talk to both of them. Next up, you will build the Express API that caches product data in Redis and publishes order events to Kafka.

Build the Express API with Redis Caching

Your infrastructure is running and both connections are verified. Now it's time to build the application layer that actually uses Redis and Kafka.

In this step, you'll create an Express API with two endpoints. The first endpoint serves product data using a cache-aside pattern with Redis. The second endpoint accepts orders and publishes events to Kafka, creating the first half of your event-driven pipeline.

In this step, get ready to:
  • Create an Express API server that imports your Redis and Kafka client modules.
  • Implement a GET /products endpoint with Redis cache-aside pattern.
  • Implement a POST /orders endpoint that publishes events to Kafka.
Create the API server file

The API server is a single file that imports the Redis and Kafka client modules you created in Step 1. It sets up Express, defines a product catalog, and connects a Kafka producer at startup.

  • Create a new file called src/api.js in VS Code.
  • Add the following setup code to src/api.js:
import express from 'express';
import crypto from 'crypto';
import redisClient from './redis-client.js';
import kafka from './kafka-client.js';

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

// Configuration constants
const PRODUCTS_CACHE_KEY = 'products:all';
const PRODUCTS_CACHE_TTL = 60;
const ORDER_TOPIC = 'order-created';

// Hardcoded product catalog (simulates a database)
const productCatalog = [
  { id: 'p1', name: 'Mechanical Keyboard', price: 89.99 },
  { id: 'p2', name: 'USB-C Hub', price: 34.99 },
  { id: 'p3', name: 'Noise-Canceling Headphones', price: 199.99 },
  { id: 'p4', name: '27-inch Monitor', price: 349.99 },
  { id: 'p5', name: 'Ergonomic Mouse', price: 59.99 },
];

// Connect Kafka producer at startup
const producer = kafka.producer();
await producer.connect();
console.log('Kafka producer connected');

What does this code do?

This sets up the foundation for your API server.

  • express and crypto are Node.js modules for HTTP routing and generating unique IDs.
  • redisClient and kafka are the connection modules you built in Step 1.
  • The productCatalog array simulates a database. In production, this would be a real database query.
  • The Kafka producer connects immediately at startup so it's ready to send messages when orders come in.
  • Save src/api.js.

You won't see a change yet. The next chunk adds the route handlers that bring the API to life.

Build the GET /products endpoint with cache-aside

The cache-aside pattern is one of the most common caching strategies in production systems. Your code checks Redis first. If the data is there (a cache hit), it returns instantly. If not (a cache miss), it fetches from the source, stores the result in Redis with a TTL, then returns it.

Why cache-aside instead of other caching patterns?

Cache-aside gives your application full control over what gets cached and when. The application decides when to read from cache, when to populate it, and the cache naturally expires stale data via the TTL.

Other patterns like write-through or write-behind add complexity that isn't needed for read-heavy endpoints like a product catalog.

  • In src/api.js, add the following GET route below the console.log('Kafka producer connected') line:
app.get('/products', async (req, res) => {
  // Check Redis cache first
  const cached = await redisClient.get(PRODUCTS_CACHE_KEY);

  if (cached) {
    console.log('Cache HIT for products');
    return res.json({ source: 'cache', products: JSON.parse(cached) });
  }

  // Cache miss - fetch from catalog and store in Redis
  console.log('Cache MISS for products');
  await redisClient.setEx(PRODUCTS_CACHE_KEY, PRODUCTS_CACHE_TTL, JSON.stringify(productCatalog));

  res.json({ source: 'catalog', products: productCatalog });
});

What does this code do?

  • redisClient.get(PRODUCTS_CACHE_KEY) checks if the product data already exists in Redis under the key products:all.
  • On a cache hit, it parses the stored JSON and returns immediately. No need to touch the source data.
  • On a cache miss, setEx stores the catalog in Redis with a 60-second TTL. After 60 seconds, Redis automatically deletes the key so the next request refreshes the data.
  • The source field in the response tells you whether the data came from cache or the catalog. This makes it easy to confirm caching is working.
  • Save src/api.js.

You won't see a change yet. The next chunk adds the POST /orders route and starts the server.

Build the POST /orders endpoint that publishes to Kafka

The orders endpoint is where the event-driven pattern kicks in. Instead of processing the order immediately, the API publishes an event to Kafka and returns right away. A separate worker process (which you'll build in the next step) picks up the event later.

  • In src/api.js, add the following code below the GET /products route's closing });:
app.post('/orders', async (req, res) => {
  const { userId, productId, quantity } = req.body;

  // Validate required fields
  if (!userId || !productId || !quantity) {
    return res.status(400).json({ error: 'userId, productId, and quantity are required' });
  }

  // Check product exists
  const product = productCatalog.find((p) => p.id === productId);
  if (!product) {
    return res.status(404).json({ error: 'Product not found' });
  }

  // Build the order event
  const event = {
    eventId: crypto.randomUUID(),
    type: 'order-created',
    userId,
    productId,
    productName: product.name,
    quantity,
    totalAmount: product.price * quantity,
    createdAt: new Date().toISOString(),
  };

  // Publish to Kafka
  await producer.send({
    topic: ORDER_TOPIC,
    messages: [{ key: userId, value: JSON.stringify(event) }],
  });

  console.log(`Order event published: ${event.eventId}`);
  res.status(202).json({ message: 'Order accepted', eventId: event.eventId });
});

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

What does this code do?

  • The handler extracts userId, productId, and quantity from the request body, then validates that all three are present.
  • crypto.randomUUID() generates a unique event ID for each order. This ID will be critical for deduplication in the worker.
  • producer.send() publishes the event to the order-created Kafka topic. The key field ensures all events from the same user go to the same partition.
  • The 202 Accepted status signals that the order was received and will be processed asynchronously. This is the standard HTTP status for async operations.
  • Save src/api.js.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/api.js and move on to testing below.

ⓧ I'd like to double check the full code

Here's the complete src/api.js file. Make sure yours matches exactly:

import express from 'express';
import crypto from 'crypto';
import redisClient from './redis-client.js';
import kafka from './kafka-client.js';

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

const PRODUCTS_CACHE_KEY = 'products:all';
const PRODUCTS_CACHE_TTL = 60;
const ORDER_TOPIC = 'order-created';

const productCatalog = [
  { id: 'p1', name: 'Mechanical Keyboard', price: 89.99 },
  { id: 'p2', name: 'USB-C Hub', price: 34.99 },
  { id: 'p3', name: 'Noise-Canceling Headphones', price: 199.99 },
  { id: 'p4', name: '27-inch Monitor', price: 349.99 },
  { id: 'p5', name: 'Ergonomic Mouse', price: 59.99 },
];

const producer = kafka.producer();
await producer.connect();
console.log('Kafka producer connected');

app.get('/products', async (req, res) => {
  const cached = await redisClient.get(PRODUCTS_CACHE_KEY);

  if (cached) {
    console.log('Cache HIT for products');
    return res.json({ source: 'cache', products: JSON.parse(cached) });
  }

  console.log('Cache MISS for products');
  await redisClient.setEx(PRODUCTS_CACHE_KEY, PRODUCTS_CACHE_TTL, JSON.stringify(productCatalog));

  res.json({ source: 'catalog', products: productCatalog });
});

app.post('/orders', async (req, res) => {
  const { userId, productId, quantity } = req.body;

  if (!userId || !productId || !quantity) {
    return res.status(400).json({ error: 'userId, productId, and quantity are required' });
  }

  const product = productCatalog.find((p) => p.id === productId);
  if (!product) {
    return res.status(404).json({ error: 'Product not found' });
  }

  const event = {
    eventId: crypto.randomUUID(),
    type: 'order-created',
    userId,
    productId,
    productName: product.name,
    quantity,
    totalAmount: product.price * quantity,
    createdAt: new Date().toISOString(),
  };

  await producer.send({
    topic: ORDER_TOPIC,
    messages: [{ key: userId, value: JSON.stringify(event) }],
  });

  console.log(`Order event published: ${event.eventId}`);
  res.status(202).json({ message: 'Order accepted', eventId: event.eventId });
});

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

Now let's verify the full pipeline is working.

  • Start the API server by running this command in your terminal:
npm run api

You should see three connection messages confirming Redis, Kafka, and the HTTP server are all ready:

Connected to Redis
Kafka producer connected
API server running on http://localhost:3000
  • Open a second terminal in VS Code by pressing Cmd+Shift+` (macOS) or Ctrl+Shift+` (Windows).
  • Test the products endpoint by running:
curl http://localhost:3000/products

The response should include "source": "catalog" because this is the first request and the cache is empty.

  • Run the same command again:
curl http://localhost:3000/products

This time the response shows "source": "cache". Redis served the data instantly without touching the catalog.

  • Test the orders endpoint by posting a new order:
curl -X POST http://localhost:3000/orders -H "Content-Type: application/json" -d '{"userId": "user1", "productId": "p1", "quantity": 2}'

You should see a 202 response with "message": "Order accepted" and an eventId.

  • Verify the event landed in Kafka by running:
docker exec -ti kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server :9092 --topic order-created --from-beginning

You should see the JSON event you just published, including the eventId, userId, and totalAmount fields. Press Ctrl+C to stop the consumer.

Not seeing the expected output?

  • If you see a connection error on startup, make sure both Docker containers are still running with docker ps. Restart them with docker start redis kafka if needed.
  • If curl returns Connection refused, the API server may not be running. Check the first terminal for errors.
  • If the Kafka consumer shows nothing, make sure you included --from-beginning in the command. Without it, the consumer only shows new messages.

Help me debug my API server connection

Why 202 Accepted instead of 200 OK?

A 200 OK means the work is done. A 202 Accepted means the server received the request and will process it later. Since the actual order processing happens in a separate worker (which you'll build next), 202 is the honest response.

This is how production systems like Stripe and Shopify handle webhook-driven workflows. Accept fast, process async.

Your API is live and producing events. Next up, you'll build the worker process that consumes these Kafka events and uses Redis to prevent duplicate processing.

Build the Kafka Consumer Worker

Your API is publishing order events into Kafka. Every time someone POSTs to /orders, a message lands in the order-created topic. But nobody is listening yet.

In this step, you will build a separate worker process that consumes those events from Kafka and processes each order exactly once. The worker uses Redis to track which events it has already seen, preventing duplicate processing even if Kafka redelivers a message.

In this step, get ready to:
  • Create a Kafka consumer with a consumer group.
  • Implement Redis-based idempotency to prevent duplicate processing.
  • Run the worker and verify exactly-once event handling.
Create the consumer with a consumer group

A consumer group allows multiple worker instances to share the load of processing messages from a topic. Kafka tracks which messages each group has consumed, so if a worker restarts, it picks up where it left off.

  • Create a new file called src/worker.js in VS Code.
  • Add the consumer setup code by pasting the following into src/worker.js:
// Import shared Redis and Kafka clients
import redisClient from './redis-client.js';
import kafka from './kafka-client.js';

// Configuration constants
const ORDER_TOPIC = 'order-created';
const CONSUMER_GROUP = 'order-processing-group';
const DEDUPE_TTL = 300;

// Create a consumer with a group ID, connect, and subscribe
const consumer = kafka.consumer({ 'group.id': CONSUMER_GROUP });
await consumer.connect();
await consumer.subscribe({ topics: [ORDER_TOPIC] });

console.log(`Worker subscribed to topic: ${ORDER_TOPIC}`);

What does this code do?

This sets up the consumer side of your pipeline.

  • kafka.consumer({ 'group.id': CONSUMER_GROUP }) creates a consumer that belongs to the order-processing-group group. Kafka uses this group ID to track consumption progress.
  • await consumer.subscribe({ topics: [ORDER_TOPIC] }) tells the consumer to listen to the order-created topic.
  • DEDUPE_TTL = 300 sets the deduplication window to 300 seconds (5 minutes). Any event processed within this window will be detected as a duplicate if redelivered.

What is a consumer group?

If you run three copies of this worker with the same group ID, Kafka splits the topic's partitions across them. Each message goes to exactly one worker in the group. If one worker crashes, Kafka reassigns its partitions to the survivors.

This is how production systems scale event processing horizontally without duplicating work.

You won't see output yet. The next chunk adds the message handler that makes the consumer actually process events.

Implement Redis-based idempotency and process orders

In distributed systems, Kafka can redeliver a message if a consumer crashes before confirming it processed the event. Without protection, this means the same order could be charged twice. Redis solves this by acting as a fast lookup for whether you have already handled a given event.

  • Add the message processing loop below the existing code in src/worker.js:
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    // Parse the incoming Kafka message
    const raw = message.value.toString();
    const event = JSON.parse(raw);
    const dedupeKey = `dedupe:${event.eventId}`;

    // Check Redis for an existing deduplication key
    const alreadyProcessed = await redisClient.get(dedupeKey);
    if (alreadyProcessed) {
      console.log(`Duplicate detected, skipping: ${event.eventId}`);
      return;
    }

    // Process the order (simulate payment)
    console.log(`Processing order: ${event.eventId}`);
    console.log(`  User: ${event.userId}`);
    console.log(`  Product: ${event.productName} x${event.quantity}`);
    console.log(`  Total: $${event.totalAmount}`);
    console.log(`  Payment charged for order ${event.eventId}`);

    // Mark this event as processed with a TTL
    await redisClient.setEx(dedupeKey, DEDUPE_TTL, '1');
    console.log(`Order ${event.eventId} processed successfully\n`);
  },
});

What does this code do?

The eachMessage callback runs once for every message Kafka delivers to this consumer.

  • It builds a deduplication key like dedupe:abc-123 from the event's unique ID, then checks Redis for that key.
  • If the key exists, the event was already processed. The worker logs "Duplicate detected" and skips it.
  • If the key does not exist, the worker processes the order, simulates payment, then stores the key in Redis with a 300-second TTL.
  • Kafka automatically commits the offset after eachMessage resolves successfully. This means the offset is only committed AFTER Redis confirms the deduplication key is stored.

Why use a TTL on the deduplication key?

Without a TTL, deduplication keys would accumulate in Redis forever. A 300-second window is long enough to catch any redelivery from Kafka (which typically retries within seconds) but short enough to avoid unbounded memory growth.

  • Save src/worker.js.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before running the worker.

ⓧ I'd like to double check the full code

Here is the complete src/worker.js file:

import redisClient from './redis-client.js';
import kafka from './kafka-client.js';

const ORDER_TOPIC = 'order-created';
const CONSUMER_GROUP = 'order-processing-group';
const DEDUPE_TTL = 300;

const consumer = kafka.consumer({ 'group.id': CONSUMER_GROUP });
await consumer.connect();
await consumer.subscribe({ topics: [ORDER_TOPIC] });

console.log(`Worker subscribed to topic: ${ORDER_TOPIC}`);

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const raw = message.value.toString();
    const event = JSON.parse(raw);
    const dedupeKey = `dedupe:${event.eventId}`;

    const alreadyProcessed = await redisClient.get(dedupeKey);
    if (alreadyProcessed) {
      console.log(`Duplicate detected, skipping: ${event.eventId}`);
      return;
    }

    console.log(`Processing order: ${event.eventId}`);
    console.log(`  User: ${event.userId}`);
    console.log(`  Product: ${event.productName} x${event.quantity}`);
    console.log(`  Total: $${event.totalAmount}`);
    console.log(`  Payment charged for order ${event.eventId}`);

    await redisClient.setEx(dedupeKey, DEDUPE_TTL, '1');
    console.log(`Order ${event.eventId} processed successfully\n`);
  },
});
Run the worker and test duplicate detection

Now you will run the worker alongside the API and watch events flow through the full pipeline. You will also prove that the Redis-based idempotency prevents duplicate processing.

  • Make sure your API is still running. If you stopped it earlier, start it again in a terminal by running:
npm run api
  • Open a second terminal window in VS Code by pressing Cmd+Shift+` (macOS) or Ctrl+Shift+` (Windows).
  • Start the worker in this new terminal by running:
npm run worker

You should see output confirming the worker connected to Redis and subscribed to the topic:

Connected to Redis
Worker subscribed to topic: order-created

Why two separate processes?

The API and the worker are decoupled through Kafka. The API does not wait for order processing to finish before responding to the client.

This separation means you can scale the worker independently, restart it without downtime on the API, and add more workers to handle spikes in order volume.

  • Open a third terminal and submit a test order by running:
curl -X POST http://localhost:3000/orders -H "Content-Type: application/json" -d '{"userId": "user1", "productId": "p1", "quantity": 2}'

In your worker terminal, you should see the order being processed:

Processing order: <some-uuid>
  User: user1
  Product: Mechanical Keyboard x2
  Total: $179.98
  Payment charged for order <some-uuid>
Order <some-uuid> processed successfully
  • Now test duplicate detection. Copy the eventId that appeared in the worker output and publish a duplicate message directly to Kafka by running:
echo '{"eventId":"PASTE_YOUR_EVENT_ID_HERE","type":"order-created","userId":"user1","productId":"p1","productName":"Mechanical Keyboard","quantity":2,"totalAmount":179.98,"createdAt":"2026-01-01T00:00:00.000Z"}' | docker exec -i kafka /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server :9092 --topic order-created

What is happening here?

You are using the Kafka console producer to inject a raw message directly into the order-created topic with the same eventId as the order you just placed. This simulates what happens in production when Kafka redelivers a message after a network hiccup or consumer crash.

In your worker terminal, you should see the duplicate get caught:

Duplicate detected, skipping: <same-uuid>
  • Verify the deduplication keys exist in Redis by running:
docker exec -it redis redis-cli KEYS "dedupe:*"

You should see at least one key matching dedupe:<your-event-id> in the output.

Worker not processing messages?

  • Make sure the API terminal shows "Kafka producer connected" before you POST an order. If not, the API failed to connect to Kafka.
  • Check that both Docker containers are running with docker ps. You should see both redis and kafka listed.
  • If you see a connection error, try restarting the containers with docker restart redis kafka and then restart both the API and worker.

Help me debug my Kafka worker

Your worker is consuming events and guaranteeing exactly-once processing. Next up, you will run the full pipeline end-to-end and stress-test the cache, message flow, and deduplication together.

Test the Full Pipeline End-to-End

You have a working API that caches products in Redis and publishes orders to Kafka. You also have a worker that consumes those events with deduplication. Now it is time to run both processes together and see the full event-driven pipeline in action.

In this step, you will launch both processes simultaneously and test every layer: Redis caching, Kafka message flow, and idempotency protection.

In this step, get ready to:
  • Run the API and worker simultaneously in separate terminals.
  • Test Redis cache hits, misses, and manual invalidation.
  • Verify idempotency by replaying events and inspecting Redis keys.
Run both processes simultaneously

Production event-driven systems have multiple processes running at the same time. The API handles incoming HTTP requests while workers process events asynchronously. You will replicate this setup locally using two terminal windows.

  • In VS Code, open a new terminal by pressing Ctrl+` (macOS and Windows).
  • Start the API server by running:
npm run api

You should see three connection messages followed by the server ready message:

Connected to Redis
Kafka producer connected
API server running on http://localhost:3000
  • Open a second terminal in VS Code by pressing Ctrl+` again (this creates a new terminal tab).
  • Start the worker by running:
npm run worker

You should see the worker connect and subscribe:

Connected to Redis
Worker subscribed to topic: order-created
  • Open a third terminal in VS Code. This will be your testing terminal where you send requests.
  • Post your first order by running:
curl -X POST http://localhost:3000/orders \
  -H "Content-Type: application/json" \
  -d '{"userId": "user1", "productId": "p1", "quantity": 2}'

You should see a JSON response with a 202 status, a confirmation message, and a unique eventId.

  • Switch to the worker terminal. You should see the order processing logs showing the user, product, quantity, total, and payment confirmation.
  • Post a few more orders with different products by running each command:
curl -X POST http://localhost:3000/orders \
  -H "Content-Type: application/json" \
  -d '{"userId": "user2", "productId": "p3", "quantity": 1}'
curl -X POST http://localhost:3000/orders \
  -H "Content-Type: application/json" \
  -d '{"userId": "user1", "productId": "p5", "quantity": 3}'

Check the worker terminal. Each order appears with its details and a "processed successfully" confirmation.

What is happening across the terminals?

Your third terminal sends HTTP requests to the API (terminal 1). The API publishes events to Kafka. The worker (terminal 2) picks up those events from Kafka and processes them. This decoupled architecture means the API responds immediately without waiting for order processing to complete.

Worker not showing any output?

Make sure the worker is running in its terminal and shows "Worker subscribed to topic: order-created". If the worker started before the API posted any messages, it may take a moment for the first message to arrive.

If you see a connection error, confirm both Docker containers are still running with docker ps.

Help me debug my worker connection

Test the Redis cache behavior

The cache-aside pattern stores data in Redis on the first request, then serves subsequent requests directly from cache. You will observe this pattern by watching the API logs as you hit the products endpoint.

  • In your testing terminal (terminal 3), fetch products for the first time by running:
curl http://localhost:3000/products
  • Check the API terminal (terminal 1). You should see Cache MISS for products in the logs.
  • Look at the JSON response. The source field says catalog, confirming the data came from the hardcoded product list.
  • Hit the same endpoint again by running the same command:
curl http://localhost:3000/products
  • Check the API terminal again. This time you should see Cache HIT for products.
  • The JSON response now shows source as cache, confirming the data came from Redis.

Why is the second response faster?

On the first request, the API checked Redis (empty), fetched from the catalog, stored the result in Redis with a 60-second TTL, and returned it. On the second request, the API found the data already in Redis and returned it immediately. This is the cache-aside pattern in action.

  • Now manually delete the Redis cache key to simulate cache expiry by running:
docker exec -it redis redis-cli DEL products:all

You should see the output (integer) 1, confirming one key was deleted.

  • Fetch products one more time by running:
curl http://localhost:3000/products
  • Check the API terminal. You should see Cache MISS for products again, proving Redis repopulated the cache from the catalog.

Still seeing cache hits after deleting the key?

Make sure you typed the key name exactly as products:all. Redis keys are case-sensitive. If the DEL command returned (integer) 0, no key was found with that name.

Help me debug my Redis cache invalidation

Test idempotency by replaying events

In production, Kafka may redeliver messages due to network issues or consumer rebalancing. The Redis-based deduplication ensures each order is processed exactly once, even if the same event arrives multiple times.

  • In your testing terminal, post an order and copy the eventId from the response by running:
curl -X POST http://localhost:3000/orders \
  -H "Content-Type: application/json" \
  -d '{"userId": "user3", "productId": "p2", "quantity": 1}'
  • Check the worker terminal. You should see the order processed successfully.
  • Copy the eventId value from the curl response (it looks like a UUID such as a1b2c3d4-e5f6-7890-abcd-ef1234567890).
  • Now replay that same event directly into Kafka using the console producer. Run the following command, replacing the eventId with the one you copied:
docker exec -ti kafka /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server :9092 --topic order-created

What is this command doing?

This opens an interactive Kafka producer connected directly to the order-created topic. Each line you type becomes a message in Kafka, bypassing the API entirely. This simulates what happens when Kafka redelivers a message the worker already processed.

  • The console producer is now waiting for input. Type a JSON message using the eventId you copied (replace the UUID below with your actual eventId):
{"eventId":"a1b2c3d4-e5f6-7890-abcd-ef1234567890","type":"order-created","userId":"user3","productId":"p2","productName":"USB-C Hub","quantity":1,"totalAmount":34.99,"createdAt":"2026-06-14T10:00:00.000Z"}
  • Press Enter to send the message, then press Ctrl+C to exit the console producer.
  • Check the worker terminal. You should see:
Duplicate detected, skipping: a1b2c3d4-e5f6-7890-abcd-ef1234567890

The worker recognized the eventId was already in Redis and skipped reprocessing.

  • Inspect the deduplication keys stored in Redis by running:
docker exec -it redis redis-cli KEYS "dedupe:*"

You should see a list of keys like dedupe:<eventId> for every order the worker processed. Each key has a 300-second TTL that automatically expires, allowing the same eventId to be reprocessed after 5 minutes if needed.

  • Verify the TTL on one of the keys by running (replace the UUID with one from your output):
docker exec -it redis redis-cli TTL "dedupe:a1b2c3d4-e5f6-7890-abcd-ef1234567890"

You should see a number between 1 and 300, representing the remaining seconds before the key expires.

Why use a TTL on deduplication keys?

Without a TTL, deduplication keys would accumulate forever and Redis memory would grow unbounded. The 300-second window is long enough to catch any Kafka redelivery (which typically happens within seconds), but short enough that Redis stays lean.

In production, you would tune this TTL based on your system's maximum redelivery window and acceptable memory usage.

Not seeing the duplicate detected message?

Make sure the eventId you pasted into the console producer matches exactly the one from your earlier curl response. Even one character difference means it is a new event to the worker.

Also confirm the JSON you typed is valid. A missing quote or bracket will cause the worker to fail parsing and you will see an error instead.

Help me troubleshoot my idempotency test

Your full pipeline is working. The API caches products in Redis, publishes orders to Kafka, and the worker processes each order exactly once thanks to Redis-based deduplication. Next up, you will extend this system with dead-letter handling and rate limiting in the Secret Mission.

Secret mission

Add Dead-Letter Handling and Rate Limiting

Your pipeline processes orders, but what happens when an order is invalid? And what stops a single user from flooding your system with requests? In production, every pipeline needs guardrails. Add a dead-letter topic for failed orders and Redis-based rate limiting to protect your API.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • Docker container redis (Redis, no persistent storage, free).
  • Docker container kafka (Apache Kafka in KRaft mode, no persistent storage, free).
  • Local order-pipeline project folder on your Desktop.

Keep everything running

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

  • Your Redis and Kafka containers continue running in the background.
  • You can restart the API and worker any time with npm run api and npm run worker.
  • Both containers use minimal memory and no paid services.

Pause - I'll come back to this later

Stop the running containers to free up memory. Your containers will persist and can be restarted later.

  • Stop both containers by running:
docker stop redis kafka
  • When you're ready to resume, restart them by running:
docker start redis kafka

What happens to my data?

Redis data (cache keys, deduplication keys, rate limit counters) will be lost when the container stops because no persistent volume was configured. Kafka topic data persists in the container's filesystem until the container is removed.

Delete - I don't want to use this again

Remove all project resources and start fresh. There is no data to preserve since both containers use ephemeral storage.

  • Remove both Docker containers by running:
docker rm -f redis kafka
  • Verify the containers are gone by running:
docker ps -a

You should no longer see redis or kafka in the list.

  • Delete the project folder by running:
rm -rf ~/Desktop/order-pipeline

On Windows, run this instead:

rmdir /s /q %USERPROFILE%\Desktop\order-pipeline
  • Confirm the folder is removed by running:
ls ~/Desktop/order-pipeline

You should see a "No such file or directory" message.

Nice Work!

Nice Work!

You just built a production-grade event-driven order processing pipeline from scratch. Two separate processes, connected through Kafka, coordinated by Redis.

You've learned how to:

  • Build an Express.js API that caches product data in Redis using the cache-aside pattern and publishes order events to Kafka.
  • Create a separate Kafka consumer worker that subscribes to topics via consumer groups and processes messages independently from the API.
  • Implement Redis-based idempotency to guarantee exactly-once processing of events across distributed processes.
  • Secret Mission: Added dead-letter topic handling for failed orders and Redis INCR-based rate limiting to protect the API from abuse.

Ready to quiz yourself?