Kafka #

Apache Kafka is a distributed event streaming platform designed to handle millions of events per second with very low latency. Unlike traditional message brokers such as RabbitMQ, which delete messages after consumption, Kafka stores all events in a log that can be replayed — very useful for audit trails, event sourcing, and stream processing. Go supports Kafka through two popular libraries: github.com/IBM/sarama (pure Go, more flexible) and github.com/confluentinc/confluent-kafka-go (a librdkafka wrapper, more performant).

Basic Kafka Concepts #

Topic      → a message category (like a "channel" or "queue name")
Partition  → a topic subdivision for parallelism; each topic has 1+ partitions
Offset     → the unique position of each message in a partition (monotonically increasing)
Producer   → an application that writes messages to a topic
Consumer   → an application that reads messages from a topic
Consumer Group → a group of consumers working together to read a topic;
               each partition is read by only one consumer in the group
Broker     → a Kafka server; a cluster consists of many brokers

Visualizing the Kafka Architecture #

Here’s a diagram of the message flow from Producers to Topics/Partitions inside the Kafka Broker, until consumed by a Consumer Group:

flowchart TD
    subgraph Producers["Sender Applications"]
        P1[Producer 1]
        P2[Producer 2]
    end

    subgraph KafkaCluster["Kafka Cluster (Topic: order-events)"]
        subgraph P0["Partition 0"]
            P0Msg1["Msg 1 (Offset 0)"]
            P0Msg2["Msg 2 (Offset 1)"]
            P0Msg3["Msg 3 (Offset 2)"]
        end
        subgraph P1Dir["Partition 1"]
            P1Msg1["Msg 1 (Offset 0)"]
            P1Msg2["Msg 2 (Offset 1)"]
        end
    end

    subgraph ConsGroup["Consumer Group (order-processor)"]
        C1[Consumer A]
        C2[Consumer B]
    end

    P1 -->|"Publish to Partition 0"| P0
    P2 -->|"Publish to Partition 1"| P1Dir

    P0 -->|"Read"| C1
    P1Dir -->|"Read"| C2

Installation & Go Library Comparison #

There are two main libraries for interacting with Kafka in Go. Here’s the comparison before installing:

Category	`github.com/IBM/sarama`	`github.com/confluentinc/confluent-kafka-go`
Dependency	Pure Go (No C/gcc compiler needed)	Cgo wrapper (requires the `librdkafka` C library)
Cross Compilation	Very easy	Quite complex (must include C build)
Feature Compliance	Kafka protocol features implemented manually	Always up to date (officially supported by the Confluent team)
Stability	Stable for common scenarios	Highly recommended for large enterprise workloads

Running the Installation #

# Sarama — pure Go, more flexible
go get github.com/IBM/sarama

# Or Confluent — needs the librdkafka C library
go get github.com/confluentinc/confluent-kafka-go/v2/kafka

Producer — Sending Messages #

The Sarama Sync Producer #

import (
    "github.com/IBM/sarama"
    "encoding/json"
)

func newSyncProducer(brokers []string) (sarama.SyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll  // wait for all replicas
    config.Producer.Retry.Max = 5
    config.Producer.Return.Successes = true

    // Idempotent producer — prevents duplicates during retries
    config.Producer.Idempotent = true
    config.Net.MaxOpenRequests = 1  // required for idempotence

    return sarama.NewSyncProducer(brokers, config)
}

func sendMessage(producer sarama.SyncProducer, topic string, key string, value interface{}) error {
    data, err := json.Marshal(value)
    if err != nil {
        return fmt.Errorf("marshal: %w", err)
    }

    msg := &sarama.ProducerMessage{
        Topic: topic,
        Key:   sarama.StringEncoder(key),
        Value: sarama.ByteEncoder(data),
        Headers: []sarama.RecordHeader{
            {Key: []byte("content-type"), Value: []byte("application/json")},
            {Key: []byte("source"), Value: []byte("myapp")},
        },
    }

    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        return fmt.Errorf("send: %w", err)
    }

    log.Printf("Message sent to partition=%d offset=%d", partition, offset)
    return nil
}

The Sarama Async Producer — High Throughput #

func newAsyncProducer(brokers []string) (sarama.AsyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForLocal
    config.Producer.Compression = sarama.CompressionSnappy
    config.Producer.Flush.Frequency = 500 * time.Millisecond
    config.Producer.Return.Successes = true
    config.Producer.Return.Errors = true

    return sarama.NewAsyncProducer(brokers, config)
}

func runAsyncProducer(producer sarama.AsyncProducer) {
    // Goroutine to handle successes and errors
    go func() {
        for {
            select {
            case msg := <-producer.Successes():
                log.Printf("OK: partition=%d offset=%d", msg.Partition, msg.Offset)
            case err := <-producer.Errors():
                log.Printf("ERROR: %v", err)
            }
        }
    }()

    // Send messages
    for i := 0; i < 100; i++ {
        producer.Input() <- &sarama.ProducerMessage{
            Topic: "events",
            Key:   sarama.StringEncoder(fmt.Sprintf("key-%d", i)),
            Value: sarama.StringEncoder(fmt.Sprintf(`{"seq":%d}`, i)),
        }
    }
}

Consumer — Reading Messages #

Consumer Groups — The Recommended Way #

Consumer groups are the idiomatic Kafka way — many consumers share the load of reading partitions:

// The Handler interface that needs to be implemented
type OrderEventHandler struct {
    db *sql.DB
}

// Setup is called when the consumer session starts
func (h *OrderEventHandler) Setup(sarama.ConsumerGroupSession) error {
    log.Println("Consumer group session started")
    return nil
}

// Cleanup is called when the session ends
func (h *OrderEventHandler) Cleanup(sarama.ConsumerGroupSession) error {
    log.Println("Consumer group session ended")
    return nil
}

// ConsumeClaim processes messages from one partition
func (h *OrderEventHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for msg := range claim.Messages() {
        // Process the message
        if err := h.processMessage(msg); err != nil {
            log.Printf("Error processing message offset=%d: %v", msg.Offset, err)
            // Don't commit if processing failed (depends on the retry strategy)
            continue
        }

        // Commit the offset — mark the message as processed
        // MarkMessage only buffers the commit, it doesn't flush immediately
        session.MarkMessage(msg, "")
    }
    return nil
}

func (h *OrderEventHandler) processMessage(msg *sarama.ConsumerMessage) error {
    log.Printf("Processing: topic=%s partition=%d offset=%d key=%s",
        msg.Topic, msg.Partition, msg.Offset, string(msg.Key))

    var event OrderEvent
    if err := json.Unmarshal(msg.Value, &event); err != nil {
        return fmt.Errorf("unmarshal: %w", err)
    }

    // Process the event
    return h.handleOrderEvent(event)
}

func (h *OrderEventHandler) handleOrderEvent(event OrderEvent) error {
    switch event.Type {
    case "order.created":
        log.Printf("New order: %s, total Rp%.0f", event.OrderID, event.Total)
        // Save to DB, send a notification, etc.
    case "order.paid":
        log.Printf("Order paid: %s", event.OrderID)
    case "order.shipped":
        log.Printf("Order shipped: %s", event.OrderID)
    default:
        log.Printf("Unknown event: %s", event.Type)
    }
    return nil
}

// Run the consumer group
func runConsumerGroup(ctx context.Context, brokers []string, groupID string, topics []string) error {
    config := sarama.NewConfig()
    config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{
        sarama.NewBalanceStrategyRoundRobin(),
    }
    config.Consumer.Offsets.Initial = sarama.OffsetNewest
    // Auto commit every 1 second (default)
    config.Consumer.Offsets.AutoCommit.Enable = true
    config.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second

    client, err := sarama.NewConsumerGroup(brokers, groupID, config)
    if err != nil {
        return fmt.Errorf("create consumer group: %w", err)
    }
    defer client.Close()

    handler := &OrderEventHandler{}

    for {
        // Consume will block until all partitions are assigned
        // It returns if a rebalance occurs — hence the loop
        if err := client.Consume(ctx, topics, handler); err != nil {
            if errors.Is(err, sarama.ErrClosedConsumerGroup) {
                return nil
            }
            return fmt.Errorf("consume: %w", err)
        }

        // Check whether the context was cancelled
        if ctx.Err() != nil {
            return nil
        }
    }
}

Delivery Semantics #

Kafka supports three levels of delivery guarantees:

AT-MOST-ONCE (fastest, can lose messages):
  Producer: acks=0 (no confirmation wait)
  Consumer: commit the offset BEFORE processing the message

AT-LEAST-ONCE (default, can duplicate):
  Producer: acks=all, retry=true
  Consumer: commit the offset AFTER successfully processing the message

EXACTLY-ONCE (safest, most complex):
  Producer: idempotent=true + transactional
  Consumer: read_committed + manual commit within a transaction

At-Least-Once with Manual Commits #

// Disable auto-commit, commit manually after processing
config.Consumer.Offsets.AutoCommit.Enable = false

func (h *Handler) ConsumeClaim(
    session sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for msg := range claim.Messages() {
        if err := h.process(msg); err != nil {
            log.Printf("FAILED to process offset=%d, will be reprocessed", msg.Offset)
            // Not marked → will be processed again on restart
            continue
        }
        // Mark only on success
        session.MarkMessage(msg, "")
        // Commit now (don't wait for the interval)
        session.Commit()
    }
    return nil
}

Confluent Kafka Go — The Performant Alternative #

import "github.com/confluentinc/confluent-kafka-go/v2/kafka"

// Producer
func newConfluentProducer(brokers string) (*kafka.Producer, error) {
    return kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers":  brokers,
        "acks":               "all",
        "retries":            5,
        "enable.idempotence": true,
    })
}

func produce(p *kafka.Producer, topic, key string, value []byte) error {
    deliveryCh := make(chan kafka.Event)
    err := p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{
            Topic:     &topic,
            Partition: kafka.PartitionAny,
        },
        Key:   []byte(key),
        Value: value,
    }, deliveryCh)
    if err != nil {
        return err
    }

    e := <-deliveryCh
    m := e.(*kafka.Message)
    if m.TopicPartition.Error != nil {
        return m.TopicPartition.Error
    }
    log.Printf("Delivered: partition=%d offset=%d",
        m.TopicPartition.Partition, m.TopicPartition.Offset)
    return nil
}

// Consumer
func newConfluentConsumer(brokers, groupID string) (*kafka.Consumer, error) {
    return kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers":        brokers,
        "group.id":                 groupID,
        "auto.offset.reset":        "earliest",
        "enable.auto.commit":       false,
        "max.poll.interval.ms":     300000,
    })
}

func consumeMessages(c *kafka.Consumer, topics []string, ctx context.Context) error {
    if err := c.SubscribeTopics(topics, nil); err != nil {
        return err
    }

    for {
        select {
        case <-ctx.Done():
            return nil
        default:
        }

        msg, err := c.ReadMessage(100 * time.Millisecond)
        if err != nil {
            if err.(kafka.Error).Code() == kafka.ErrTimedOut {
                continue
            }
            return err
        }

        log.Printf("Received: %s [%d] @ %d\n",
            *msg.TopicPartition.Topic, msg.TopicPartition.Partition,
            msg.TopicPartition.Offset)

        // Process the message
        if err := processMessage(msg.Value); err != nil {
            log.Printf("Error: %v", err)
            continue
        }

        // Manual commit
        c.CommitMessage(msg)
    }
}

Complete Example Program — Event-Driven Order System #

package main

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

    "github.com/IBM/sarama"
)

const (
    brokerAddr  = "localhost:9092"
    topicOrders = "orders"
    groupID     = "order-processor"
)

type OrderEvent struct {
    EventID    string    `json:"event_id"`
    Type       string    `json:"type"`
    OrderID    string    `json:"order_id"`
    CustomerID string    `json:"customer_id"`
    Total      float64   `json:"total"`
    Items      []string  `json:"items"`
    Timestamp  time.Time `json:"timestamp"`
}

// ── Producer ──────────────────────────────────────────────────

type EventProducer struct {
    producer sarama.SyncProducer
}

func NewEventProducer(brokers []string) (*EventProducer, error) {
    cfg := sarama.NewConfig()
    cfg.Producer.RequiredAcks = sarama.WaitForAll
    cfg.Producer.Retry.Max = 5
    cfg.Producer.Return.Successes = true
    cfg.Producer.Idempotent = true
    cfg.Net.MaxOpenRequests = 1

    p, err := sarama.NewSyncProducer(brokers, cfg)
    if err != nil {
        return nil, err
    }
    return &EventProducer{producer: p}, nil
}

func (ep *EventProducer) Publish(event OrderEvent) error {
    data, err := json.Marshal(event)
    if err != nil {
        return err
    }

    part, offset, err := ep.producer.SendMessage(&sarama.ProducerMessage{
        Topic:     topicOrders,
        Key:       sarama.StringEncoder(event.OrderID),  // same key → same partition
        Value:     sarama.ByteEncoder(data),
        Timestamp: event.Timestamp,
    })
    if err != nil {
        return fmt.Errorf("publish: %w", err)
    }
    log.Printf("[PRODUCER] event=%s order=%s → partition=%d offset=%d",
        event.Type, event.OrderID, part, offset)
    return nil
}

func (ep *EventProducer) Close() error {
    return ep.producer.Close()
}

// ── Consumer ──────────────────────────────────────────────────

type orderHandler struct {
    processed int
}

func (h *orderHandler) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (h *orderHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }

func (h *orderHandler) ConsumeClaim(
    sess sarama.ConsumerGroupSession,
    claim sarama.ConsumerGroupClaim,
) error {
    for msg := range claim.Messages() {
        var event OrderEvent
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            log.Printf("[CONSUMER] ERROR unmarshal offset=%d: %v", msg.Offset, err)
            sess.MarkMessage(msg, "")
            continue
        }

        log.Printf("[CONSUMER] partition=%d offset=%d type=%s order=%s",
            msg.Partition, msg.Offset, event.Type, event.OrderID)

        // Simulate processing
        switch event.Type {
        case "order.created":
            fmt.Printf("  → Save order %s to DB (total Rp%.0f)\n",
                event.OrderID, event.Total)
        case "order.paid":
            fmt.Printf("  → Mark order %s as PAID\n", event.OrderID)
        case "order.shipped":
            fmt.Printf("  → Update tracking for order %s\n", event.OrderID)
        }

        h.processed++
        sess.MarkMessage(msg, "")
    }
    return nil
}

// ── Main ──────────────────────────────────────────────────────

func main() {
    brokers := []string{brokerAddr}
    ctx, cancel := context.WithCancel(context.Background())

    // Capture signals for graceful shutdown
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

    // Start the consumer in a separate goroutine
    cfg := sarama.NewConfig()
    cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
    cfg.Consumer.Offsets.AutoCommit.Enable = true

    group, err := sarama.NewConsumerGroup(brokers, groupID, cfg)
    if err != nil {
        log.Fatal("Consumer group:", err)
    }

    handler := &orderHandler{}
    go func() {
        for {
            if err := group.Consume(ctx, []string{topicOrders}, handler); err != nil {
                if errors.Is(err, sarama.ErrClosedConsumerGroup) {
                    return
                }
                log.Printf("Consume error: %v", err)
            }
            if ctx.Err() != nil {
                return
            }
        }
    }()

    // Producer: send several events
    producer, err := NewEventProducer(brokers)
    if err != nil {
        log.Fatal("Producer:", err)
    }

    events := []OrderEvent{
        {
            EventID: "evt-001", Type: "order.created",
            OrderID: "ORD-2024-001", CustomerID: "CUST-42",
            Total: 1_500_000, Items: []string{"Laptop", "Mouse"},
            Timestamp: time.Now(),
        },
        {
            EventID: "evt-002", Type: "order.paid",
            OrderID: "ORD-2024-001", CustomerID: "CUST-42",
            Total: 1_500_000, Timestamp: time.Now().Add(5 * time.Second),
        },
        {
            EventID: "evt-003", Type: "order.created",
            OrderID: "ORD-2024-002", CustomerID: "CUST-99",
            Total: 350_000, Items: []string{"Keyboard"},
            Timestamp: time.Now().Add(10 * time.Second),
        },
        {
            EventID: "evt-004", Type: "order.shipped",
            OrderID: "ORD-2024-001", CustomerID: "CUST-42",
            Total: 1_500_000, Timestamp: time.Now().Add(15 * time.Second),
        },
    }

    fmt.Println("=== Sending Events to Kafka ===")
    for _, evt := range events {
        if err := producer.Publish(evt); err != nil {
            log.Printf("Failed to publish: %v", err)
        }
        time.Sleep(200 * time.Millisecond)
    }

    // Wait for the consumer to process or a stop signal
    select {
    case <-sigCh:
        fmt.Println("\nReceived shutdown signal...")
    case <-time.After(5 * time.Second):
        fmt.Printf("\nTimeout. The consumer processed %d messages.\n", handler.processed)
    }

    cancel()
    producer.Close()
    group.Close()
    fmt.Println("Done.")
}

Summary #

Kafka stores messages in a replayable log — unlike RabbitMQ, which deletes them after consumption.
Consumer groups are the idiomatic way — many consumers share partitions; one partition is read by only one consumer at a time.
The same key always goes to the same partition — important for preserving event order per entity (order, user).
The client.Consume loop must run because it returns whenever a rebalance happens.
Manual commits (AutoCommit.Enable = false) for safe at-least-once delivery — commit only after successful processing.
Idempotent producers (Idempotent = true) prevent duplicates during retries.
Async producers for high throughput — handle success/errors via separate channels.
Compression (Snappy, LZ4, Zstd) to reduce bandwidth — very useful for large payloads.
OffsetOldest to read from the beginning; OffsetNewest for only new messages.
Graceful shutdown: cancel the context → wait for the consumer to finish → close the producer → close the consumer group.

← Previous: Elasticsearch Next: RabbitMQ →