Kafka vs RabbitMQ vs NATS vs SQS: Choosing the Right Message Broker

package main
 
import (
	"context"
	"encoding/json"
	"fmt"
	"time"
 
	"github.com/segmentio/kafka-go"
)
 
type OrderEvent struct {
	OrderID   string    `json:"order_id"`
	Amount    int64     `json:"amount_cents"`
	Timestamp time.Time `json:"timestamp"`
}
 
func publishOrderEvent(ctx context.Context, w *kafka.Writer, event OrderEvent) error {
	payload, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("marshal event: %w", err)
	}
	msg := kafka.Message{
		Key:   []byte(event.OrderID), // same order → same partition → ordered
		Value: payload,
	}
	return w.WriteMessages(ctx, msg)
}
 
func newKafkaWriter(brokers []string, topic string) *kafka.Writer {
	return &kafka.Writer{
		Addr:         kafka.TCP(brokers...),
		Topic:        topic,
		Balancer:     &kafka.Hash{},    // partition by key
		RequiredAcks: kafka.RequireAll, // wait for all ISR replicas
		MaxAttempts:  5,
		Async:        false, // synchronous: fail immediately on error
	}
}

func consumeOrders(ctx context.Context, brokers []string, topic, groupID string) error {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:        brokers,
		Topic:          topic,
		GroupID:        groupID,
		CommitInterval: 0, // default: CommitMessages commits synchronously
		StartOffset:    kafka.LastOffset,
	})
	defer r.Close()
 
	for {
		msg, err := r.FetchMessage(ctx) // does NOT auto-commit (ReadMessage does)
		if err != nil {
			return fmt.Errorf("fetch message: %w", err) // ctx cancelled or reader closed
		}
 
		if err := processOrder(ctx, msg); err != nil {
			// Don't skip ahead: committing any later offset on this partition
			// commits this one too. Stop; a restart resumes from the last commit.
			return fmt.Errorf("process offset %d: %w", msg.Offset, err)
		}
 
		// Commit only after successful processing
		if err := r.CommitMessages(ctx, msg); err != nil {
			return fmt.Errorf("commit offset: %w", err)
		}
	}
}

package main
 
import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"
 
	amqp "github.com/rabbitmq/amqp091-go"
)
 
type TaskPayload struct {
	TaskID  string `json:"task_id"`
	Type    string `json:"type"`
	Payload []byte `json:"payload"`
}
 
func publishTask(ch *amqp.Channel, exchange, routingKey string, task TaskPayload) error {
	// Enable publisher confirms
	if err := ch.Confirm(false); err != nil {
		return fmt.Errorf("enable confirms: %w", err)
	}
 
	body, err := json.Marshal(task)
	if err != nil {
		return fmt.Errorf("marshal task: %w", err)
	}
 
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
 
	confirmation, err := ch.PublishWithDeferredConfirmWithContext(
		ctx, exchange, routingKey, true, false,
		amqp.Publishing{
			DeliveryMode: amqp.Persistent, // survive broker restart
			ContentType:  "application/json",
			MessageId:    task.TaskID,
			Body:         body,
		},
	)
	if err != nil {
		return fmt.Errorf("publish: %w", err) // else confirmation is nil → Wait() panics
	}
 
	acked, err := confirmation.WaitContext(ctx) // wait for broker ack
	if err != nil {
		return fmt.Errorf("wait for confirm: %w", err)
	}
	if !acked {
		return fmt.Errorf("broker nacked task %s", task.TaskID)
	}
	return nil
}

func consumeFromQueue(ch *amqp.Channel, queue string) error {
	msgs, err := ch.Consume(
		queue, "", false, false, false, false, nil, // autoAck=false: manual ack only
	)
	if err != nil {
		return fmt.Errorf("consume: %w", err)
	}
 
	for msg := range msgs {
		if err := processTask(msg.Body); err != nil {
			log.Printf("process task: %v — requeueing", err)
			// Negative ack: message goes back to queue
			if nerr := msg.Nack(false, true); nerr != nil {
				return fmt.Errorf("nack: %w", nerr)
			}
			continue
		}
		// Positive ack: message removed
		if err := msg.Ack(false); err != nil {
			return fmt.Errorf("ack: %w", err)
		}
	}
	return fmt.Errorf("consume %s: delivery channel closed", queue) // channel or connection lost
}

package main
 
import (
	"fmt"
	"log"
	"time"
 
	"github.com/nats-io/nats.go"
)
 
func rpcExample(nc *nats.Conn) error {
	// Service: subscribe and respond
	if _, err := nc.Subscribe("api.users.get", func(msg *nats.Msg) {
		response := `{"user_id": "42", "name": "Alice"}`
		if err := msg.Respond([]byte(response)); err != nil {
			log.Printf("respond: %v", err)
		}
	}); err != nil {
		return fmt.Errorf("subscribe: %w", err)
	}
 
	// Client: request with timeout
	reply, err := nc.Request("api.users.get", []byte("42"), 2*time.Second)
	if err != nil {
		return err // timeout or no responder
	}
 
	log.Printf("response: %s", string(reply.Data))
	return nil
}

func jetStreamPublish(js nats.JetStreamContext, subject string) error {
	_, err := js.Publish(subject, []byte(`{"order_id":"123"}`))
	return err
}
 
func jetStreamConsume(js nats.JetStreamContext) error {
	// Create stream
	if _, err := js.AddStream(&nats.StreamConfig{
		Name:     "ORDERS",
		Subjects: []string{"orders.>"},
		Storage:  nats.FileStorage,
		MaxAge:   7 * 24 * time.Hour,
	}); err != nil {
		return fmt.Errorf("add stream: %w", err)
	}
 
	// Create durable consumer
	if _, err := js.AddConsumer("ORDERS", &nats.ConsumerConfig{
		Durable:       "order-processor",
		AckPolicy:     nats.AckExplicitPolicy,
		AckWait:       30 * time.Second,
		FilterSubject: "orders.created",
	}); err != nil {
		return fmt.Errorf("add consumer: %w", err)
	}
 
	// Consume
	sub, err := js.PullSubscribe("orders.created", "order-processor")
	if err != nil {
		return fmt.Errorf("subscribe: %w", err)
	}
	msgs, err := sub.Fetch(10)
	if err != nil {
		return fmt.Errorf("fetch: %w", err)
	}
	for _, msg := range msgs {
		if err := msg.Ack(); err != nil {
			log.Printf("ack failed for msg: %v", err)
		}
	}
	return nil
}

package main
 
import (
	"context"
	"encoding/json"
	"fmt"
	"log"
 
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
)
 
type EmailTask struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
	Body    string `json:"body"`
}
 
func sendToSQS(ctx context.Context, queueURL string, task EmailTask) error {
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		return fmt.Errorf("load AWS config: %w", err)
	}
	client := sqs.NewFromConfig(cfg)
 
	body, err := json.Marshal(task)
	if err != nil {
		return fmt.Errorf("marshal task: %w", err)
	}
	_, err = client.SendMessage(ctx, &sqs.SendMessageInput{
		QueueUrl:    aws.String(queueURL),
		MessageBody: aws.String(string(body)),
	})
	return err
}
 
func pollSQS(ctx context.Context, queueURL string) error {
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		return fmt.Errorf("load AWS config: %w", err)
	}
	client := sqs.NewFromConfig(cfg)
 
	result, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(queueURL),
		MaxNumberOfMessages: 10,
		WaitTimeSeconds:     20, // long polling
		VisibilityTimeout:   60, // 60s to process
	})
	if err != nil {
		return fmt.Errorf("receive: %w", err) // else result is nil → panic below
	}
 
	for _, msg := range result.Messages {
		var task EmailTask
		if err := json.Unmarshal([]byte(*msg.Body), &task); err != nil {
			log.Printf("unmarshal task: %v — leaving for redrive, a malformed body won't fix itself on retry", err)
			continue // visibility timeout will requeue; a DLQ redrive policy stops this from looping forever
		}
 
		if err := sendEmail(ctx, task); err != nil {
			log.Printf("send email (msg %s): %v — leaving for redelivery", aws.ToString(msg.MessageId), err)
			continue // visibility timeout will requeue
		}
 
		if _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
			QueueUrl:      aws.String(queueURL),
			ReceiptHandle: msg.ReceiptHandle,
		}); err != nil {
			log.Printf("delete message: %v — email already sent, expect a duplicate redelivery", err)
		}
	}
	return nil
}

# docker-compose.brokers.yml — paste alongside your test runner
services:
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
      # Tight retention for tests so disk doesn't bloat across runs.
      KAFKA_LOG_RETENTION_HOURS: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
 
  rabbitmq:
    image: rabbitmq:3.13-management
    ports: ["5672:5672", "15672:15672"]
    environment:
      RABBITMQ_DEFAULT_USER: test
      RABBITMQ_DEFAULT_PASS: test
 
  nats:
    image: nats:2.10
    command: ["-js", "-sd", "/data"]
    ports: ["4222:4222"]
 
  localstack:                         # SQS-compatible
    image: localstack/localstack:3.4
    ports: ["4566:4566"]
    environment:
      SERVICES: sqs
      DEFAULT_REGION: us-east-1

package consumer
 
import (
	"context"
	"errors"
	"fmt"
 
	"github.com/twmb/franz-go/pkg/kgo"
)
 
func Consume(ctx context.Context, brokers []string, group, topic string,
	process func(context.Context, *kgo.Record) error) error {
	cl, err := kgo.NewClient(
		kgo.SeedBrokers(brokers...),
		kgo.ConsumerGroup(group),
		kgo.ConsumeTopics(topic),
		kgo.DisableAutoCommit(),
		kgo.BlockRebalanceOnPoll(), // no partition moves between poll and AllowRebalance
	)
	if err != nil {
		return fmt.Errorf("new client: %w", err)
	}
	defer cl.CloseAllowingRebalance()
 
	for {
		fetches := cl.PollRecords(ctx, 100) // small batches finish inside the rebalance timeout
		if err := ctx.Err(); err != nil {
			return err
		}
		if fetches.IsClientClosed() {
			return errors.New("kafka client closed")
		}
		var fetchErr error
		fetches.EachError(func(t string, p int32, err error) {
			fetchErr = errors.Join(fetchErr, fmt.Errorf("fetch %s[%d]: %w", t, p, err))
		})
		if fetchErr != nil {
			return fetchErr
		}
 
		records := fetches.Records()
		for _, rec := range records {
			if err := process(ctx, rec); err != nil {
				// Commit nothing from this batch; whoever owns the partition next re-reads it.
				return fmt.Errorf("process %s[%d]@%d: %w", rec.Topic, rec.Partition, rec.Offset, err)
			}
		}
		if err := cl.CommitRecords(ctx, records...); err != nil {
			return fmt.Errorf("commit: %w", err)
		}
		cl.AllowRebalance() // only now can partitions move to another member
	}
}

# Replicate retail/orders across three regions. Assumes the clusters are registered
# and the "retail" tenant is allowed on all three.
bin/pulsar-admin namespaces create retail/orders
bin/pulsar-admin namespaces set-clusters retail/orders \
  --clusters us-east-1,eu-west-1,ap-southeast-1
bin/pulsar-admin namespaces set-retention retail/orders --size 100G --time 7d
# The broker rejects a backlog quota at or above the retention size.
bin/pulsar-admin namespaces set-backlog-quota retail/orders \
  --limit 50G --policy producer_request_hold   # hold producers, don't drop
bin/pulsar-admin namespaces set-schema-compatibility-strategy retail/orders \
  --compatibility BACKWARD_TRANSITIVE

// Redpanda is Kafka API compatible — the same kafka-go client works,
// you just swap brokers. Compare latency histograms before migrating.
package main
 
import (
	"context"
	"time"
 
	"github.com/segmentio/kafka-go"
)
 
func newRedpandaWriter(brokers []string, topic string) *kafka.Writer {
	return &kafka.Writer{
		Addr:         kafka.TCP(brokers...),
		Topic:        topic,
		Balancer:     &kafka.Hash{},
		RequiredAcks: kafka.RequireAll,
		// Tight client-side timeout: fail fast rather than queue behind a slow ack.
		WriteTimeout: 50 * time.Millisecond,
		BatchTimeout: 5 * time.Millisecond, // tight batching for low latency
		Async:        false,
	}
}
 
func publishLowLatency(ctx context.Context, w *kafka.Writer, key, value []byte) error {
	deadline, cancel := context.WithTimeout(ctx, 20*time.Millisecond)
	defer cancel()
	return w.WriteMessages(deadline, kafka.Message{Key: key, Value: value})
}

# Provision a per-tenant stream with bounded retention and ACLs in three commands.
# Run this from your tenant-onboarding job; it is idempotent.
TENANT_ID="$1"
 
nats stream add "tenant-${TENANT_ID}" \
  --subjects="tenants.${TENANT_ID}.events.>" \
  --storage=file \
  --retention=limits \
  --max-age=720h \
  --max-bytes=10737418240 \
  --replicas=3 \
  --discard=old
 
nats consumer add "tenant-${TENANT_ID}" "tenant-${TENANT_ID}-webhooks" \
  --filter="tenants.${TENANT_ID}.events.>" \
  --ack=explicit \
  --max-deliver=10 \
  --wait=30s \
  --backoff=linear --backoff-min=1s --backoff-max=5m
 
nats auth user add "tenant-${TENANT_ID}-publisher" \
  --pub="tenants.${TENANT_ID}.events.>" \
  --sub="_INBOX.>" \
  --account=tenants