Secure Transaction Pipeline with Kafka

Build a fraud-scoring pipeline with Go, Kafka, Redis, PostgreSQL, and Docker.

Introduction

30 Second Summary

Every time you tap your card or hit "pay" online, a chain of backend systems evaluates that transaction in milliseconds. These systems communicate by sending events to each other, scoring each payment for fraud before it clears.

In this project, you will build a secure transaction pipeline with Go and Apache Kafka. Two microservices communicate through events to ingest, score, and store payment transactions with fraud detection.

What You'll Build

A secure transaction processing system built on event-driven architecture, with two Go microservices communicating through Apache Kafka, backed by PostgreSQL for persistence and Redis for idempotent fraud scoring.

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

  • A Docker Compose stack running Kafka, PostgreSQL, and Redis locally.
  • A Go transaction producer service with an HTTP API that publishes payment events to a Kafka topic.
  • A Go fraud detection consumer with Redis-backed idempotency that scores and persists verified transactions to PostgreSQL.
  • Secret Mission: Deploy the full stack to AWS EC2 free tier with Docker Compose.

Are there any prerequisites?

This project assumes familiarity with Go basics like structs, interfaces, goroutines, and HTTP handlers. You should also have a basic understanding of Docker Compose and SQL.

Set Up Your Development Environment

Before you can build an event-driven pipeline, you need the infrastructure that makes it possible. Three services will power your system: Kafka for messaging, PostgreSQL for storage, and Redis for caching.

In this step, you will create the project folder, spin up all three services using Docker Compose, and install the Go libraries your pipeline depends on.

In this step, get ready to:
  • Create the project directory and initialize the Go module.
  • Define and start the Docker Compose infrastructure stack.
  • Install Go dependencies for Kafka, PostgreSQL, and Redis.
Create the project directory and initialize the Go module

Every Go project starts with a module. The module defines the project's import path and tracks its dependencies.

  • In VS Code, open the integrated terminal by pressing Ctrl+` (Linux/Windows) or Cmd+` (macOS).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, move into it, and initialize the Go module by running these commands:
mkdir transaction-pipeline
cd transaction-pipeline
go mod init transaction-pipeline
  • Confirm the folder and module file exist by running:
ls

You should see go.mod in the output.

What does go mod init do?

The go mod init command creates a go.mod file that declares your project's module path and tracks all of its dependencies.

The name transaction-pipeline becomes the root import path for all packages in this project.

🙋‍♀️ Seeing 'command not found: go'?

Make sure Go is installed and your PATH includes the Go binary location. Run go version to confirm it returns 1.25 or higher.

help me troubleshoot Go not being found in my PATH on Linux

  • Open the transaction-pipeline folder in VS Code by clicking File in the top menu bar.
  • Click Open Folder.
  • Navigate to Desktop, select the transaction-pipeline folder, and click Open.
  • Open a new terminal in VS Code by pressing Ctrl+` (Linux/Windows) or Cmd+` (macOS).

The terminal now opens directly inside your transaction-pipeline folder.

Create the Docker Compose file and start the stack

Your pipeline needs three infrastructure services running locally. Docker Compose lets you define and start all of them with a single command.

  • In VS Code's Explorer sidebar, click the new file icon next to TRANSACTION-PIPELINE and name it docker-compose.yml.
  • Paste the following configuration into docker-compose.yml:
services:
  kafka:
    image: apache/kafka:4.3.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3

  postgres:
    image: postgres:16
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: orders
      POSTGRES_PASSWORD: orders
      POSTGRES_DB: orders

  redis:
    image: redis:7
    container_name: redis
    ports:
      - "6379:6379"

What does this configuration do?

  • Kafka runs in KRaft mode (no ZooKeeper needed). It acts as the message broker between your producer and consumer services. The KAFKA_ADVERTISED_LISTENERS setting tells your Go code where to connect.
  • PostgreSQL stores confirmed orders. The user, password, and database are all set to orders for simplicity.
  • Redis tracks which orders have already been processed. This prevents duplicates when a consumer restarts and re-reads messages.

💡 Why KRaft mode instead of ZooKeeper?

Kafka 4.x removed ZooKeeper entirely and uses KRaft (Kafka Raft) for internal cluster coordination. This means one fewer container to manage and a simpler local setup.

  • Save docker-compose.yml.
  • Start the infrastructure stack by running this command in your terminal:
docker compose up -d

Docker pulls the images and starts all three containers in the background. This may take a minute on the first run.

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

You should see three containers listed: kafka, postgres, and redis. Each should show a status of Up.

Seeing a port conflict error?

If a container fails to start because a port is already in use, another process is occupying that port. Stop the conflicting process or change the host port in docker-compose.yml. Common conflicts are PostgreSQL on port 5432 and Redis on port 6379.

help me resolve a Docker Compose port conflict

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml and all three containers show Up in docker compose ps.

ⓧ I'd like to double check the full code

services:
  kafka:
    image: apache/kafka:4.3.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3

  postgres:
    image: postgres:16
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: orders
      POSTGRES_PASSWORD: orders
      POSTGRES_DB: orders

  redis:
    image: redis:7
    container_name: redis
    ports:
      - "6379:6379"
Install Go dependencies and verify Kafka

Your Go services will use four external libraries. Installing them now adds them to your go.mod file so Go can resolve imports when you write the producer and consumer.

  • Install all project dependencies by running these commands:
go get github.com/twmb/franz-go/pkg/kgo@v1.21.3
go get github.com/google/uuid
go get github.com/lib/pq
go get github.com/redis/go-redis/v9

What are these libraries?

  • franz-go is a pure-Go Kafka client. It handles producing and consuming messages. You pin it to v1.21.3 for reproducibility.
  • uuid generates unique order IDs.
  • lib/pq is the PostgreSQL driver for Go's database/sql package.
  • go-redis is the Redis client for Go. You will use it for idempotency checks.
  • Verify Kafka is ready to accept connections by running this command:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

The command should return without error. You may see internal Kafka topics listed, or an empty result. Both are fine.

What does this command do?

This runs the Kafka CLI tool inside the kafka container and asks it to list all topics on the broker. If it returns without error, your Go code can connect to the same address (localhost:9092).

🙋‍♀️ Seeing a connection refused error?

Kafka can take 10-15 seconds to finish starting. Wait a moment and try the command again. If it still fails, check that the kafka container is running with docker compose ps.

help me troubleshoot Kafka connection issues in Docker

Your infrastructure is running and your dependencies are locked. Next up, you will build the producer service that publishes order events into Kafka.

Build the Order Producer Service

Build the Transaction Producer Service

Your infrastructure is up and running. Now it's time to build the first microservice that puts it to work.

The producer service is an HTTP API that accepts order requests and publishes them as events to a Kafka topic. This is the entry point of the pipeline. Every order that comes in through the API becomes a message that downstream services can consume independently.

In this step, get ready to:
  • Create the producer service with an HTTP server and data model.
  • Implement the order creation handler that publishes events to Kafka.
  • Test the full flow from HTTP request to Kafka message.
Create the producer service file

The producer needs a data model (what an order looks like), a Kafka client (how to send messages), and an HTTP server (how to receive requests). You will set all of these up in a single Go file.

  • Create the producer directory and main.go file inside it by running these commands:
mkdir producer
touch producer/main.go
  • Confirm the files exist by running:
ls producer/

You should see main.go in the output.

  • Open producer/main.go in VS Code and paste the following code to set up the package, imports, data model, and server:
package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/google/uuid"
	"github.com/twmb/franz-go/pkg/kgo"
)

// Order represents a confirmed order event flowing through Kafka
type Order struct {
	ID        string    `json:"id"`
	Customer  string    `json:"customer"`
	Product   string    `json:"product"`
	Quantity  int       `json:"quantity"`
	Price     float64   `json:"price"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

// OrderRequest is the incoming payload from the HTTP client
type OrderRequest struct {
	Customer string  `json:"customer"`
	Product  string  `json:"product"`
	Quantity int     `json:"quantity"`
	Price    float64 `json:"price"`
}

// Global Kafka client shared by all HTTP handlers
var client *kgo.Client

func main() {
	var err error
	// Connect to Kafka using the franz-go client
	client, err = kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
	)
	if err != nil {
		log.Fatalf("failed to create kafka client: %v", err)
	}
	defer client.Close()

	// Register HTTP routes
	http.HandleFunc("/orders", createOrder)
	http.HandleFunc("/health", healthCheck)

	fmt.Println("Producer running on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

// healthCheck responds with a simple OK for liveness probes
func healthCheck(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("ok"))
}

What does this code do?

This sets up the skeleton of your producer service.

  • The Order struct defines the full shape of an order event (what Kafka will carry). The OrderRequest struct defines only what the HTTP client needs to send. Your service fills in the rest (ID, status, timestamp).
  • kgo.NewClient creates a connection to your local Kafka broker. The SeedBrokers option tells franz-go where to find Kafka.
  • The main function registers two routes and starts the HTTP server on port 8080.

The file will not compile yet because createOrder is referenced but not defined. You will add it next.

Implement the order creation handler

The createOrder function is where the real work happens. It decodes an incoming HTTP request, generates a unique ID, and publishes the order as a Kafka message.

  • In producer/main.go, paste the following function below healthCheck:
func createOrder(w http.ResponseWriter, r *http.Request) {
	// Only accept POST requests
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	// Decode the incoming JSON body into an OrderRequest
	var req OrderRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "invalid request body", http.StatusBadRequest)
		return
	}

	// Build the full Order with a generated UUID and timestamp
	order := Order{
		ID:        uuid.New().String(),
		Customer:  req.Customer,
		Product:   req.Product,
		Quantity:  req.Quantity,
		Price:     req.Price,
		Status:    "placed",
		CreatedAt: time.Now(),
	}

	// Serialize the order to JSON for the Kafka message value
	value, err := json.Marshal(order)
	if err != nil {
		http.Error(w, "failed to marshal order", http.StatusInternalServerError)
		return
	}

	// Create a Kafka record with the order ID as the key
	// Using the order ID as key ensures all events for one order land on the same partition
	record := &kgo.Record{
		Topic: "orders",
		Key:   []byte(order.ID),
		Value: value,
	}

	// Produce synchronously and wait for Kafka acknowledgment
	if err := client.ProduceSync(r.Context(), record).FirstErr(); err != nil {
		http.Error(w, fmt.Sprintf("failed to produce: %v", err), http.StatusInternalServerError)
		return
	}

	// Return the created order to the client
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(order)
}

What does this code do?

This handler does four things in sequence: validate the request, build the order, publish to Kafka, and respond to the client.

  • ProduceSync sends the record to Kafka and blocks until the broker acknowledges it. This means the HTTP response only returns 201 if the message was safely stored in Kafka.
  • Using the order ID as the record key guarantees that all events for the same order land on the same Kafka partition. This preserves ordering per order.
  • Kafka auto-creates the orders topic when the first message arrives. No manual topic creation needed.
  • Save producer/main.go.

Seeing a compile error?

Make sure the createOrder function is outside the main function (not nested inside it). Each function in Go lives at the package level.

Also check that your import block includes all the packages listed in the first code chunk. A missing import like github.com/google/uuid causes an unresolved reference error.

Help me fix a compile error in my producer/main.go file

✔️ Awesome, I've got everything!

Great. Double check you have saved producer/main.go before moving on.

ⓧ I'd like to double check the full code

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/google/uuid"
	"github.com/twmb/franz-go/pkg/kgo"
)

type Order struct {
	ID        string    `json:"id"`
	Customer  string    `json:"customer"`
	Product   string    `json:"product"`
	Quantity  int       `json:"quantity"`
	Price     float64   `json:"price"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

type OrderRequest struct {
	Customer string  `json:"customer"`
	Product  string  `json:"product"`
	Quantity int     `json:"quantity"`
	Price    float64 `json:"price"`
}

var client *kgo.Client

func main() {
	var err error
	client, err = kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
	)
	if err != nil {
		log.Fatalf("failed to create kafka client: %v", err)
	}
	defer client.Close()

	http.HandleFunc("/orders", createOrder)
	http.HandleFunc("/health", healthCheck)

	fmt.Println("Producer running on :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

func createOrder(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}

	var req OrderRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "invalid request body", http.StatusBadRequest)
		return
	}

	order := Order{
		ID:        uuid.New().String(),
		Customer:  req.Customer,
		Product:   req.Product,
		Quantity:  req.Quantity,
		Price:     req.Price,
		Status:    "placed",
		CreatedAt: time.Now(),
	}

	value, err := json.Marshal(order)
	if err != nil {
		http.Error(w, "failed to marshal order", http.StatusInternalServerError)
		return
	}

	record := &kgo.Record{
		Topic: "orders",
		Key:   []byte(order.ID),
		Value: value,
	}

	if err := client.ProduceSync(r.Context(), record).FirstErr(); err != nil {
		http.Error(w, fmt.Sprintf("failed to produce: %v", err), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(order)
}

func healthCheck(w http.ResponseWriter, r *http.Request) {
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("ok"))
}
Test the producer end to end

With the code complete, you can now run the producer and send it a real HTTP request. You will confirm that the order flows all the way into Kafka.

  • Start the producer service by running:
go run ./producer

You should see Producer running on :8080 printed in the terminal. The server is now listening for requests.

Seeing a connection error on startup?

If the producer logs a Kafka connection error, your Docker Compose stack may not be running. Switch to another terminal and run docker compose ps to verify all three containers are healthy. If they are not, run docker compose up -d to restart them.

Help me troubleshoot my Kafka producer connection error

  • Open a second terminal. Send a test order to the producer by running:
curl -s -X POST http://localhost:8080/orders \
  -H "Content-Type: application/json" \
  -d '{"customer":"Alice","product":"Mechanical Keyboard","quantity":1,"price":89.99}'

You should see a JSON response with a 201 status containing your order. It includes a generated UUID in the id field and status set to placed.

  • Verify the message reached Kafka by reading from the orders topic:
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning --max-messages 1

You should see the same JSON order object printed in your terminal. This confirms the message traveled from your HTTP API into Kafka successfully.

Why use the console consumer?

The Kafka console consumer is a built-in CLI tool that reads messages directly from a topic. The --from-beginning flag tells it to start from the earliest offset. The --max-messages 1 flag makes it exit after printing one message so your terminal does not hang.

  • Press Ctrl+C in the first terminal to stop the producer when you are done testing.

Curl returned an error or no output?

  • If you see Connection refused, the producer is not running. Check the first terminal for error messages and restart with go run ./producer.
  • If you see method not allowed, make sure you included -X POST in your curl command.
  • If the Kafka console consumer shows nothing, wait a few seconds and try again. Kafka may need a moment to create the topic on first use.

Help me debug my producer test

Your producer service is live. It accepts orders over HTTP and reliably delivers them to Kafka. Next up, you will build the consumer that reads these events and processes them.

Build the Order Consumer Service

Build the Fraud Detection Consumer

Your producer is publishing order events to Kafka. Now you need a service on the other side that reads those events and does something with them.

In this step, you will build the consumer service. It connects to Kafka as part of a consumer group, reads order messages from the orders topic, and handles graceful shutdown so no messages are lost when you stop the service.

In this step, get ready to:
  • Create the consumer service with a Kafka client configured for consumer group membership.
  • Implement the consume loop that reads and processes order events.
  • Test consumer group offset tracking by restarting the consumer.
Create the consumer service file

The consumer is a separate Go program that lives in its own directory, just like the producer. It connects to the same Kafka cluster but uses a different client configuration to read messages instead of writing them.

  • Create the consumer directory inside your transaction-pipeline folder by running this command:
mkdir consumer
  • Create consumer/main.go and add the following code:
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
)

type Order struct {
	ID        string    `json:"id"`
	Customer  string    `json:"customer"`
	Product   string    `json:"product"`
	Quantity  int       `json:"quantity"`
	Price     float64   `json:"price"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

func main() {
	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.ConsumerGroup("order-processors"),
		kgo.ConsumeTopics("orders"),
	)
	if err != nil {
		log.Fatalf("failed to create kafka client: %v", err)
	}
	defer client.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigCh
		fmt.Println("\nShutting down...")
		cancel()
	}()

	fmt.Println("Consumer running, waiting for orders...")

	for {
		fetches := client.PollFetches(ctx)
		if ctx.Err() != nil {
			break
		}
		if errs := fetches.Errors(); len(errs) > 0 {
			for _, e := range errs {
				log.Printf("fetch error: topic=%s partition=%d err=%v", e.Topic, e.Partition, e.Err)
			}
			continue
		}

		iter := fetches.RecordIter()
		for !iter.Done() {
			record := iter.Next()
			if err := processRecord(ctx, record); err != nil {
				log.Printf("failed to process record: %v", err)
			}
		}
	}
}

func processRecord(ctx context.Context, record *kgo.Record) error {
	var order Order
	if err := json.Unmarshal(record.Value, &order); err != nil {
		return fmt.Errorf("unmarshal error: %w", err)
	}
	log.Printf("received order: id=%s customer=%s product=%s", order.ID, order.Customer, order.Product)
	return nil
}

What does this code do?

The client initialization uses two key options that make this a consumer instead of a producer:

  • kgo.ConsumerGroup("order-processors") joins this consumer to a named group. Kafka tracks which messages the group has already processed. If this consumer crashes and restarts, it picks up where it left off.
  • kgo.ConsumeTopics("orders") tells the client which topic to read from.

The signal handling block listens for Ctrl+C (SIGINT) or a termination signal (SIGTERM). When either arrives, it cancels the context, which causes PollFetches to stop blocking and the for loop to exit cleanly.

  • Save consumer/main.go.
  • Verify it compiles by running:
go build ./consumer

If the command completes with no output, the build succeeded.

Seeing a compile error?

  • If you see an error about missing packages, run go mod tidy to resolve dependencies and try again.
  • If you see a syntax error, check that all braces and parentheses are balanced. The go func() { ... }() goroutine needs both the closing brace and the trailing ().

Still stuck? Help me fix a compile error in my consumer/main.go file.

Understand the consume loop

The consume loop is the heart of your consumer service. It continuously polls Kafka for new messages and routes each one through your processing function.

How does the consume loop work?

  • client.PollFetches(ctx) blocks until Kafka delivers one or more records. If the context is cancelled (from Ctrl+C), it returns immediately so you can exit the loop.
  • fetches.Errors() checks for any transport-level problems (network issues, broker failures). The loop logs them and moves on rather than crashing.
  • fetches.RecordIter() gives you an iterator over every record in the fetch batch. Each call to iter.Next() returns the next kgo.Record.
  • processRecord unmarshals the JSON value back into an Order struct and logs it. For now it only logs. In the next step, you will add database persistence and idempotency.

💡 What is a consumer group?

A consumer group lets multiple consumer instances share the work of reading from a topic. Kafka assigns each partition to exactly one member of the group, and tracks the last processed offset per partition.

When your consumer restarts, Kafka tells it which offset to resume from. This means already-processed messages are not re-delivered. You will verify this behaviour in the next substep.

Test the consumer with live orders

Time to see the full producer-consumer flow in action. You will run the consumer in one terminal and send orders from another.

  • Open a new terminal tab or window. Start the consumer by running:
go run ./consumer

You should see the message Consumer running, waiting for orders... and the process stays alive waiting for messages.

  • Open a second terminal tab. Start the producer by running:
go run ./producer
  • In a third terminal tab, send a test order by running:
curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d '{"customer":"Alice","product":"Keyboard","quantity":1,"price":75}'

Switch back to the consumer terminal. You should see a log line like:

received order: id=<uuid> customer=Alice product=Keyboard

What should I see?

The consumer prints a log line for each order it receives, showing the order ID, customer name, and product. The ID will be a UUID generated by the producer.

  • Now test consumer group offset tracking. Stop the consumer with Ctrl+C.
  • Send another order while the consumer is stopped:
curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d '{"customer":"Bob","product":"Mouse","quantity":2,"price":30}'
  • Restart the consumer by running:
go run ./consumer

You should see only Bob's order logged. The Alice order from earlier does not reappear because Kafka tracked the consumer group's offset.

Why does only the new order appear?

When your consumer processes a record, franz-go automatically commits the offset back to Kafka. On restart, Kafka tells the consumer to start reading from the next uncommitted offset. This is the core value of consumer groups: crash recovery without reprocessing.

🙋‍♀️ Consumer not receiving orders?

  • Make sure your Docker containers are still running. Check with docker compose ps and look for Kafka in a healthy state.
  • Confirm the producer is running and returned a 201 response when you sent the curl request.
  • If the consumer logs fetch error messages, Kafka may still be starting up. Wait a few seconds and try again.

Need more help? Help me debug why my consumer is not receiving Kafka messages.

✔️ Awesome, I've got everything!

Great. Double check that you have saved consumer/main.go and that your consumer logs orders in real time.

ⓧ I'd like to double check the full code

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
)

type Order struct {
	ID        string    `json:"id"`
	Customer  string    `json:"customer"`
	Product   string    `json:"product"`
	Quantity  int       `json:"quantity"`
	Price     float64   `json:"price"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

func main() {
	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.ConsumerGroup("order-processors"),
		kgo.ConsumeTopics("orders"),
	)
	if err != nil {
		log.Fatalf("failed to create kafka client: %v", err)
	}
	defer client.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigCh
		fmt.Println("\nShutting down...")
		cancel()
	}()

	fmt.Println("Consumer running, waiting for orders...")

	for {
		fetches := client.PollFetches(ctx)
		if ctx.Err() != nil {
			break
		}
		if errs := fetches.Errors(); len(errs) > 0 {
			for _, e := range errs {
				log.Printf("fetch error: topic=%s partition=%d err=%v", e.Topic, e.Partition, e.Err)
			}
			continue
		}

		iter := fetches.RecordIter()
		for !iter.Done() {
			record := iter.Next()
			if err := processRecord(ctx, record); err != nil {
				log.Printf("failed to process record: %v", err)
			}
		}
	}
}

func processRecord(ctx context.Context, record *kgo.Record) error {
	var order Order
	if err := json.Unmarshal(record.Value, &order); err != nil {
		return fmt.Errorf("unmarshal error: %w", err)
	}
	log.Printf("received order: id=%s customer=%s product=%s", order.ID, order.Customer, order.Product)
	return nil
}

Your consumer reads events from Kafka and proves that consumer group offsets survive restarts. Next up, you will add PostgreSQL persistence and Redis-backed idempotency so that processed orders are stored safely and never duplicated.

Add PostgreSQL Persistence and Redis Idempotency

Your consumer can read orders from Kafka, but they disappear as soon as you stop the process. Real order pipelines need durable storage and a guarantee that the same order is never processed twice.

In this step, you will connect the consumer to PostgreSQL for permanent storage and Redis for idempotency checks. Every order that flows through the pipeline will land in the database exactly once.

In this step, get ready to:
  • Connect the consumer to PostgreSQL and create the orders table on startup.
  • Add Redis-backed idempotency to prevent duplicate order processing.
  • Test that orders persist correctly and duplicates are caught.
Connect to PostgreSQL and Redis

The consumer needs two new connections: one to PostgreSQL (where confirmed orders live permanently) and one to Redis (where you track which orders have already been processed).

  • In consumer/main.go, replace the entire import block and the code above func main() with the following updated imports, Order struct, and package-level variables:
package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/twmb/franz-go/pkg/kgo"
	_ "github.com/lib/pq"
)

// Order represents an order event flowing through the pipeline
type Order struct {
	ID        string    `json:"id"`
	Customer  string    `json:"customer"`
	Product   string    `json:"product"`
	Quantity  int       `json:"quantity"`
	Price     float64   `json:"price"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

// Package-level database and cache clients
var (
	db  *sql.DB
	rdb *redis.Client
)

What changed in the imports?

You added three new imports. "database/sql" is Go's standard database interface. "github.com/redis/go-redis/v9" is the Redis client library. _ "github.com/lib/pq" is a PostgreSQL driver that registers itself with database/sql at import time (the blank identifier _ means you only need its side effects).

The var block declares db and rdb at the package level so both main() and processRecord() can access them.

Save consumer/main.go after making this change. The file will not compile yet because you still need to update main().

  • Now replace the func main() body. Find the opening of main() and replace it entirely with the following code that adds database and Redis initialization before the Kafka client setup:
func main() {
	var err error

	// Connect to PostgreSQL
	db, err = sql.Open("postgres", "host=localhost port=5432 user=orders password=orders dbname=orders sslmode=disable")
	if err != nil {
		log.Fatalf("failed to connect to postgres: %v", err)
	}
	defer db.Close()

	// Create the orders table if it doesn't exist
	if err := createTable(); err != nil {
		log.Fatalf("failed to create table: %v", err)
	}

	// Connect to Redis for idempotency tracking
	rdb = redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})
	defer rdb.Close()

	// Connect to Kafka as a consumer group member
	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.ConsumerGroup("order-processors"),
		kgo.ConsumeTopics("orders"),
	)
	if err != nil {
		log.Fatalf("failed to create kafka client: %v", err)
	}
	defer client.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Graceful shutdown on SIGINT/SIGTERM
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigCh
		fmt.Println("\nShutting down...")
		cancel()
	}()

	fmt.Println("Consumer running, waiting for orders...")

	for {
		fetches := client.PollFetches(ctx)
		if ctx.Err() != nil {
			break
		}
		if errs := fetches.Errors(); len(errs) > 0 {
			for _, e := range errs {
				log.Printf("fetch error: topic=%s partition=%d err=%v", e.Topic, e.Partition, e.Err)
			}
			continue
		}

		iter := fetches.RecordIter()
		for !iter.Done() {
			record := iter.Next()
			if err := processRecord(ctx, record); err != nil {
				log.Printf("failed to process record: %v", err)
			}
		}
	}
}

What does the new initialization do?

  • sql.Open("postgres", ...) creates a connection pool to PostgreSQL using the credentials from your Docker Compose stack.
  • createTable() ensures the orders table exists before any records are processed.
  • redis.NewClient(&redis.Options{Addr: "localhost:6379"}) connects to the Redis container for fast key lookups.
  • Add the createTable function at the bottom of consumer/main.go, after the existing processRecord function:
func createTable() error {
	_, err := db.Exec(`
		CREATE TABLE IF NOT EXISTS orders (
			id VARCHAR(36) PRIMARY KEY,
			customer VARCHAR(255) NOT NULL,
			product VARCHAR(255) NOT NULL,
			quantity INTEGER NOT NULL,
			price DECIMAL(10,2) NOT NULL,
			status VARCHAR(50) NOT NULL,
			created_at TIMESTAMP NOT NULL
		)
	`)
	return err
}

Why create the table in code?

Running the CREATE TABLE IF NOT EXISTS statement on startup means the consumer is self-contained. You do not need a separate migration step or manual SQL. The id column is the primary key, which also serves as a safety net for duplicates at the database level.

  • Save consumer/main.go.

Seeing compile errors?

  • Make sure the createTable function is outside main(). It should be a standalone function at the file level.
  • Check that your import block includes "database/sql" and _ "github.com/lib/pq" exactly as shown. The underscore before the import is required.
  • If you see an error about missing packages, run go mod tidy in your transaction-pipeline folder to resolve dependencies.

Why is my consumer not compiling?

Implement idempotent order processing

The idempotency pattern works like a checklist. Before processing any order, the consumer checks Redis: "Have I seen this order ID before?" If yes, skip it. If no, insert it into PostgreSQL and mark it as processed in Redis.

Why use both Redis and PostgreSQL for deduplication?

Redis gives you a fast in-memory lookup (microseconds) to catch duplicates before hitting the database. The ON CONFLICT DO NOTHING clause on the PostgreSQL INSERT acts as a safety net for edge cases where the Redis key expired (after 24 hours) but the order already exists in the database.

This two-layer approach is the standard production pattern: a fast cache for the hot path and a durable store as the final authority.

  • In consumer/main.go, replace the existing processRecord function with the following idempotent version:
func processRecord(ctx context.Context, record *kgo.Record) error {
	var order Order
	if err := json.Unmarshal(record.Value, &order); err != nil {
		return fmt.Errorf("unmarshal error: %w", err)
	}

	// Check Redis: has this order already been processed?
	processed, err := rdb.Get(ctx, "processed:"+order.ID).Result()
	if err == nil && processed == "true" {
		log.Printf("order %s already processed, skipping", order.ID)
		return nil
	}

	// Mark order as confirmed before persisting
	order.Status = "confirmed"

	// Insert into PostgreSQL with ON CONFLICT as a safety net
	_, err = db.ExecContext(ctx,
		`INSERT INTO orders (id, customer, product, quantity, price, status, created_at)
		 VALUES ($1, $2, $3, $4, $5, $6, $7)
		 ON CONFLICT (id) DO NOTHING`,
		order.ID, order.Customer, order.Product, order.Quantity, order.Price, order.Status, order.CreatedAt,
	)
	if err != nil {
		return fmt.Errorf("db insert error: %w", err)
	}

	// Mark this order ID as processed in Redis with 24-hour TTL
	rdb.Set(ctx, "processed:"+order.ID, "true", 24*time.Hour)

	log.Printf("processed order: id=%s customer=%s product=%s", order.ID, order.Customer, order.Product)
	return nil
}

What does this function do?

  • rdb.Get(ctx, "processed:"+order.ID) checks if this order ID has already been processed. If the key exists with value "true", the function returns early without touching PostgreSQL.
  • order.Status = "confirmed" upgrades the order status from "placed" (set by the producer) to "confirmed" (the consumer's responsibility).
  • The INSERT ... ON CONFLICT (id) DO NOTHING ensures the database never stores a duplicate row, even if the Redis key has expired.
  • rdb.Set(ctx, "processed:"+order.ID, "true", 24*time.Hour) records that processing succeeded. The 24-hour TTL means old keys clean themselves up automatically.
  • Save consumer/main.go.

Seeing an undefined error for rdb or db?

  • Make sure you have the var (db *sql.DB; rdb *redis.Client) block declared at the package level (outside any function).
  • Check that processRecord is a standalone function, not nested inside main().

Help me fix undefined variable errors in my consumer.

✔️ Awesome, I've got everything!

Great. Double check you have saved consumer/main.go and move on to testing.

ⓧ I'd like to double check the full code

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/redis/go-redis/v9"
	"github.com/twmb/franz-go/pkg/kgo"
	_ "github.com/lib/pq"
)

type Order struct {
	ID        string    `json:"id"`
	Customer  string    `json:"customer"`
	Product   string    `json:"product"`
	Quantity  int       `json:"quantity"`
	Price     float64   `json:"price"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
}

var (
	db  *sql.DB
	rdb *redis.Client
)

func main() {
	var err error

	db, err = sql.Open("postgres", "host=localhost port=5432 user=orders password=orders dbname=orders sslmode=disable")
	if err != nil {
		log.Fatalf("failed to connect to postgres: %v", err)
	}
	defer db.Close()

	if err := createTable(); err != nil {
		log.Fatalf("failed to create table: %v", err)
	}

	rdb = redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})
	defer rdb.Close()

	client, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.ConsumerGroup("order-processors"),
		kgo.ConsumeTopics("orders"),
	)
	if err != nil {
		log.Fatalf("failed to create kafka client: %v", err)
	}
	defer client.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigCh
		fmt.Println("\nShutting down...")
		cancel()
	}()

	fmt.Println("Consumer running, waiting for orders...")

	for {
		fetches := client.PollFetches(ctx)
		if ctx.Err() != nil {
			break
		}
		if errs := fetches.Errors(); len(errs) > 0 {
			for _, e := range errs {
				log.Printf("fetch error: topic=%s partition=%d err=%v", e.Topic, e.Partition, e.Err)
			}
			continue
		}

		iter := fetches.RecordIter()
		for !iter.Done() {
			record := iter.Next()
			if err := processRecord(ctx, record); err != nil {
				log.Printf("failed to process record: %v", err)
			}
		}
	}
}

func processRecord(ctx context.Context, record *kgo.Record) error {
	var order Order
	if err := json.Unmarshal(record.Value, &order); err != nil {
		return fmt.Errorf("unmarshal error: %w", err)
	}

	processed, err := rdb.Get(ctx, "processed:"+order.ID).Result()
	if err == nil && processed == "true" {
		log.Printf("order %s already processed, skipping", order.ID)
		return nil
	}

	order.Status = "confirmed"

	_, err = db.ExecContext(ctx,
		`INSERT INTO orders (id, customer, product, quantity, price, status, created_at)
		 VALUES ($1, $2, $3, $4, $5, $6, $7)
		 ON CONFLICT (id) DO NOTHING`,
		order.ID, order.Customer, order.Product, order.Quantity, order.Price, order.Status, order.CreatedAt,
	)
	if err != nil {
		return fmt.Errorf("db insert error: %w", err)
	}

	rdb.Set(ctx, "processed:"+order.ID, "true", 24*time.Hour)

	log.Printf("processed order: id=%s customer=%s product=%s", order.ID, order.Customer, order.Product)
	return nil
}

func createTable() error {
	_, err := db.Exec(`
		CREATE TABLE IF NOT EXISTS orders (
			id VARCHAR(36) PRIMARY KEY,
			customer VARCHAR(255) NOT NULL,
			product VARCHAR(255) NOT NULL,
			quantity INTEGER NOT NULL,
			price DECIMAL(10,2) NOT NULL,
			status VARCHAR(50) NOT NULL,
			created_at TIMESTAMP NOT NULL
		)
	`)
	return err
}
Test the idempotency pipeline

Now you will prove that orders persist in PostgreSQL and that duplicate processing is caught by Redis.

  • Start the consumer by running this command in a terminal:
go run ./consumer
  • In a second terminal, start the producer:
go run ./producer
  • In a third terminal, send a test order by running:
curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d '{"customer":"alice","product":"keyboard","quantity":1,"price":75}'
  • In the consumer terminal, you should see a log line like processed order: id=... customer=alice product=keyboard.
  • Verify the order landed in PostgreSQL by running:
docker exec postgres psql -U orders -c "SELECT * FROM orders"

You should see a row with status confirmed, your customer name alice, and the product keyboard.

What does the query output look like?

The psql output shows a table with columns matching your schema: id, customer, product, quantity, price, status, and created_at. The status column reads confirmed because the consumer upgraded it from placed before inserting.

Now test that idempotency actually works.

  • Stop the consumer with Ctrl+C.
  • To simulate reprocessing, temporarily change the consumer group name. In consumer/main.go, find this line:
kgo.ConsumerGroup("order-processors"),
  • Change "order-processors" to "order-processors-test" so the line looks like this:
kgo.ConsumerGroup("order-processors-test"),
  • Save and restart the consumer:
go run ./consumer
  • The consumer re-reads the same messages (because the new group has no committed offsets). In the consumer logs, you should see output like order <id> already processed, skipping.
  • Query PostgreSQL again to confirm no duplicate rows were created:
docker exec postgres psql -U orders -c "SELECT count(*) FROM orders"

The count should match the number of unique orders you sent. Redis caught the duplicate and prevented a second insert.

  • Stop the consumer, revert the group name back to "order-processors", and save the file.

Not seeing orders in PostgreSQL?

  • Make sure your Docker containers are still running. Run docker compose ps to check.
  • Verify the consumer started without errors. If you see failed to connect to postgres, confirm the postgres container is healthy and listening on port 5432.
  • If you see failed to create table, check that your connection string matches host=localhost port=5432 user=orders password=orders dbname=orders sslmode=disable exactly.

Help me debug why orders are not appearing in PostgreSQL.

Your pipeline now guarantees exactly-once processing semantics. Orders flow from the HTTP API through Kafka into PostgreSQL, and Redis makes sure no order is ever processed twice. Next up, you will stress-test the full pipeline and intentionally break things to see how each component recovers.

Test the Full Pipeline and Break Things

You have a producer that publishes orders to Kafka and a consumer that processes them into PostgreSQL with Redis-backed idempotency. But how does it all hold up under pressure?

In this step, you will stress-test the pipeline by sending a burst of orders, then intentionally break components to see how Kafka handles failures and recovery.

In this step, get ready to:
  • Send 10 orders rapidly and verify they all persist in PostgreSQL.
  • Test consumer downtime and observe Kafka's message durability.
  • Break Kafka itself and watch how the producer handles broker unavailability.
Send a burst of orders through the pipeline

To properly stress-test an event-driven system, you need to send multiple messages in quick succession. A bash loop lets you fire off 10 unique orders in under a second.

  • Make sure your Docker Compose stack is running. If you stopped it earlier, start it again by running:
docker compose up -d
  • Start the consumer in one terminal by running:
go run ./consumer
  • Open a second terminal and start the producer by running:
go run ./producer
  • Open a third terminal. Send 10 orders rapidly by running this bash loop:
for i in $(seq 1 10); do curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d "{\"customer\":\"user$i\",\"product\":\"item$i\",\"quantity\":$i,\"price\":$((i*10))}" ; done

What does this command do?

The for loop iterates from 1 to 10, sending a unique POST request each time.

Each order has a different customer (user1 through user10), product (item1 through item10), and incrementing quantity and price.

  • Verify all 10 orders made it to PostgreSQL by running:
docker exec postgres psql -U orders -c "SELECT * FROM orders"

You should see 10 rows in the output, each with status confirmed.

Don't see 10 rows?

  • Check your consumer terminal for errors. If you see connection errors to PostgreSQL or Redis, make sure the Docker Compose stack is running with docker compose ps.
  • If you had leftover rows from previous testing, you can count just the new ones. The 10 orders will have customer names user1 through user10.
  • Help me troubleshoot missing orders.
Test consumer downtime and message durability

In production, consumers go down for deployments, crashes, or scaling events. Kafka's job is to hold onto messages until the consumer comes back.

  • In the terminal running your consumer, press Ctrl+C to stop it.
  • With the consumer stopped, send 5 more orders by running:
for i in $(seq 11 15); do curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d "{\"customer\":\"user$i\",\"product\":\"item$i\",\"quantity\":$i,\"price\":$((i*10))}" ; done
  • Verify the orders are queued in Kafka by checking the consumer group lag:
docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group order-processors

What should I see?

The output shows columns including CURRENT-OFFSET, LOG-END-OFFSET, and LAG.

The LAG column shows how many messages are waiting to be consumed. You should see a total lag of 5 across the partitions. This proves Kafka is holding the messages safely even though no consumer is reading them.

  • Restart the consumer by running:
go run ./consumer
  • Watch the consumer terminal. You should see it immediately process the 5 queued orders and log each one.
  • Confirm all 15 orders now exist in PostgreSQL by running:
docker exec postgres psql -U orders -c "SELECT count(*) FROM orders"

You should see a count of 15 (or more if you had prior test orders).

Why does the consumer resume where it left off?

The order-processors consumer group tracks the last committed offset for each partition. When the consumer restarts, franz-go asks Kafka for the stored offset and picks up right where it stopped.

This is the core value of consumer groups. No messages are lost, and no messages are reprocessed (unless the consumer crashed before committing).

🙋‍♀️ Consumer not catching up?

  • If the consumer starts but doesn't process the backlog, make sure you used the same consumer group name (order-processors). A different group name means a fresh offset, which might skip already-committed messages.
  • If you restarted too quickly and see rebalancing messages, wait a few seconds for the group to stabilize.
  • Help me debug consumer group offset behavior.
Break Kafka and observe producer failure handling

The ultimate resilience test is stopping the message broker itself. This reveals how your producer handles the scenario where Kafka is completely unreachable.

  • Make sure the producer is still running in its terminal.
  • Stop the Kafka container by running:
docker stop kafka
  • Try sending an order while Kafka is down:
curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d '{"customer":"test-user","product":"test-item","quantity":1,"price":10}'

You should see an error response from the producer. The ProduceSync call returns an error because it cannot reach the broker, and your handler surfaces that as an HTTP 500 response.

Why is this important?

Because the producer uses synchronous production (ProduceSync), it knows immediately when a message fails to deliver. The HTTP client gets an honest error instead of a false success.

In production, you would add retry logic or queue failed messages locally. For now, the key takeaway is that franz-go does not silently drop messages.

  • Restart Kafka by running:
docker start kafka
  • Wait about 10 seconds for Kafka to fully start, then send another order:
curl -s -X POST http://localhost:8080/orders -H "Content-Type: application/json" -d '{"customer":"recovery-user","product":"recovery-item","quantity":1,"price":25}'

You should see a successful 201 response with the order JSON. The system has fully recovered.

Still getting errors after restarting Kafka?

  • Kafka takes a few seconds to initialize after docker start. Wait 15-20 seconds and try again.
  • If the producer process itself crashed while Kafka was down, restart it with go run ./producer.
  • Help me debug the Kafka recovery process.

Your pipeline is battle-tested. You have proven that Kafka durably holds messages during consumer downtime, that the consumer catches up automatically, and that the producer fails gracefully when the broker is unavailable. Next up, take on the Secret Mission to handle poison messages with a dead letter queue.

Secret mission

Deploy to AWS Free Tier with Docker Compose

Your pipeline runs locally on Docker Compose. But what if you want it running in the cloud, accessible from anywhere? In this secret mission, deploy your entire stack to an AWS EC2 instance on the free tier, learning real-world cloud deployment without spending a cent.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • Docker containers: Kafka, PostgreSQL, and Redis (running locally, consuming system memory and CPU).
  • Local project files: the transaction-pipeline folder on your Desktop.

Keep everything running

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

  • Your Docker containers will keep running in the background.
  • The Kafka topics, PostgreSQL data, and Redis keys all persist as long as the containers are up.
  • You can send more orders or restart the producer and consumer services at any time.

Pause - I'll come back to this later

Stop the containers to free up system memory without losing your data. Resume later with a single command.

  • Open a terminal and navigate to your transaction-pipeline folder.
  • Stop the containers by running:
docker compose stop
  • When you're ready to resume, start the containers again by running:
docker compose start

Your Kafka topics, PostgreSQL data, and Redis keys will all still be there when you come back.

Delete - I don't want to use this again

Remove all containers, data volumes, and project files to start completely fresh.

  • Open a terminal and navigate to your transaction-pipeline folder.
  • Remove all containers and their data volumes by running:
docker compose down -v
  • Navigate back to your Desktop and delete the project folder by running:
cd ~/Desktop
rm -rf transaction-pipeline

This removes all Docker containers, Kafka data, PostgreSQL tables, Redis keys, and your local Go source files.

Nice Work!

Nice Work!

You just built a production-grade event-driven order processing pipeline from scratch using Go and Kafka. This is the exact architecture pattern powering order systems at companies processing millions of transactions daily.

You learned how to:

  • Built a Go HTTP producer that publishes order events to Kafka using franz-go, with UUID-keyed records for per-order message ordering.
  • Built a Go consumer service with consumer group membership that polls and processes events, with graceful shutdown handling via OS signals.
  • Implemented Redis-backed idempotency to prevent duplicate order processing, with PostgreSQL ON CONFLICT DO NOTHING as a safety net.
  • Tested pipeline resilience by intentionally stopping components and observing how Kafka durability and consumer group offset tracking enable automatic recovery.
  • Secret Mission: Implemented a dead letter queue pattern that routes poison messages to a separate topic with error headers, preventing bad data from blocking the consumer.

Ready to quiz yourself?