Burrow

Concurrent Kafka consumer library for Go with at-least-once delivery guarantees

Overview

Burrow is a Go library that enables concurrent processing of Kafka messages while maintaining at-least-once delivery guarantees. It solves the challenge of processing I/O-heavy operations in parallel without losing messages due to out-of-order completion.

The Problem

Sequential Processing:  Slow ❌
Message 1 → [50ms] → Message 2 → [50ms] → Message 3 → [50ms]
Throughput: 20 msgs/sec

Naive Parallel: Fast but loses messages ❌
Messages [1,2,3,4,5] → Process concurrently → Complete [1,3,5,2,4]
Commit offset 5 → Lose messages 2,3,4 on restart!

Burrow: Fast AND safe ✅
Messages [1,2,3,4,5] → Process concurrently → Gap detection → Commit offset 1
→ Restart → Reprocess [2,3,4,5]

The Solution

Burrow provides:

100x throughput improvement for I/O-bound operations
At-least-once delivery through ordered offset commits
Gap detection to prevent message loss
Simple API that works like standard Kafka consumer

Installation

go get github.com/vlab-research/burrow

Quick Start

package main

import (
    "context"
    "log"
    "time"

    "github.com/confluentinc/confluent-kafka-go/v2/kafka"
    "github.com/vlab-research/burrow"
    "go.uber.org/zap"
)

func main() {
    // 1. Create standard Kafka consumer
    consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers":  "localhost:9092",
        "group.id":           "my-group",
        "auto.offset.reset":  "earliest",
        "enable.auto.commit": false, // IMPORTANT: Burrow handles commits
    })
    if err != nil {
        log.Fatal(err)
    }
    defer consumer.Close()

    // 2. Subscribe to topics
    consumer.SubscribeTopics([]string{"my-topic"}, nil)

    // 3. Create Burrow pool
    logger, _ := zap.NewProduction()
    config := burrow.DefaultConfig(logger)
    config.NumWorkers = 100 // 100 concurrent workers

    pool, err := burrow.NewPool(consumer, config)
    if err != nil {
        log.Fatal(err)
    }

    // 4. Define processing function
    processFunc := func(ctx context.Context, msg *kafka.Message) error {
        // Your I/O-heavy logic here (API calls, database queries, etc.)
        time.Sleep(50 * time.Millisecond) // Simulated I/O
        log.Printf("Processed: %s", string(msg.Value))
        return nil
    }

    // 5. Run the pool
    ctx := context.Background()
    if err := pool.Run(ctx, processFunc); err != nil {
        log.Fatal(err)
    }
}

Features

Core Capabilities

✅ Concurrent Processing - Configurable worker pool (1 to 1000+ workers)
✅ At-Least-Once Delivery - Gap detection ensures no message loss
✅ Ordered Commits - Only commits contiguous offsets
✅ Backpressure - Bounded channels prevent memory exhaustion
✅ Partition-Aware - Independent tracking per partition

Reliability

✅ Error Threshold - Automatic halt when errors exceed limit
✅ Retry with Backoff - Exponential backoff for transient failures
✅ Graceful Shutdown - Waits for inflight messages
✅ Rebalance Safe - Proper partition reassignment handling
✅ Memory Safe - Automatic cleanup after commits

Observability

✅ Metrics - GetStats() for monitoring
✅ Structured Logging - zap logger integration
✅ Context Support - Standard Go cancellation

Configuration

Basic Configuration

config := burrow.DefaultConfig(logger)

Default values:

NumWorkers: 10
JobQueueSize: 1000
ResultQueueSize: 1000
CommitInterval: 5 seconds
CommitBatchSize: 1000 messages
OnError: FatalOnError (fast failure)
ShutdownTimeout: 30 seconds

Custom Configuration

config := burrow.Config{
    NumWorkers:      100,               // 100 concurrent workers
    JobQueueSize:    10000,             // Larger buffer
    ResultQueueSize: 10000,
    CommitInterval:  10 * time.Second,  // Commit every 10s
    CommitBatchSize: 5000,              // Or every 5000 messages
    OnError:         burrow.FreezeOnError, // Use freeze mode for debugging
    ShutdownTimeout: 60 * time.Second,  // Wait longer for inflight
    Logger:          logger,
    EnableMetrics:   true,
}

pool, err := burrow.NewPool(consumer, config)

Tuning Workers

Choose worker count based on workload:

// I/O-bound work (API calls, database queries)
config.NumWorkers = 100  // High concurrency

// CPU-bound work
config.NumWorkers = runtime.NumCPU()  // Match CPU cores

// Mixed workload
config.NumWorkers = runtime.NumCPU() * 2  // Start here and measure

Monitoring

Get Runtime Statistics

stats := pool.GetStats()
fmt.Printf("Processed: %d\n", stats.MessagesProcessed)
fmt.Printf("Failed: %d\n", stats.MessagesFailed)
fmt.Printf("Committed: %d\n", stats.OffsetsCommitted)
fmt.Printf("Workers: %d\n", stats.WorkersActive)
fmt.Printf("Queued: %d\n", stats.JobsQueued)

Key Metrics to Monitor

MessagesProcessed - Throughput indicator
MessagesFailed - Error rate
OffsetsCommitted - Commit frequency
JobsQueued - Backpressure indicator (should be < QueueSize)

Alert when:

MessagesFailed is high
JobsQueued approaches JobQueueSize (backpressure)
MessagesProcessed drops unexpectedly

Error Handling

Error Behavior

When your processFunc returns an error, it signals something is seriously wrong (database down, critical service unavailable, etc.). Burrow provides two behaviors:

FatalOnError (Default) - Fast Failure:

config.OnError = burrow.FatalOnError

Immediately exits process (os.Exit(1)) on first error
No graceful shutdown, no waiting for inflight messages
Container runtime restarts with crash loop backoff
Best for production with auto-restart (Kubernetes, systemd)

FreezeOnError - Debug Mode:

config.OnError = burrow.FreezeOnError

Stops polling new messages
Waits for inflight messages to complete
Commits offsets up to the failed message
Freezes forever (requires manual restart)
Best for investigation - preserves state for debugging

Retry Logic

Burrow does NOT handle retries. Your app should handle retries in processFunc:

processFunc := func(ctx context.Context, msg *kafka.Message) error {
    // App handles retries
    for attempt := 0; attempt < 3; attempt++ {
        err := doWork(msg)
        if err == nil {
            return nil  // Success
        }

        // App decides what's retriable
        if !isRetriable(err) {
            sendToDLQ(msg, err)
            return nil  // Handled, don't error
        }

        // App controls backoff
        time.Sleep(time.Duration(1<<attempt) * time.Second)
    }

    // Exhausted retries
    sendToDLQ(msg, err)
    return nil  // We handled it
}

Return error only when something is seriously wrong (requires restart).

Return nil when you've handled the message (even if processing "failed" - e.g., sent to DLQ).

Graceful Shutdown

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

// Handle signals
go func() {
    sigch := make(chan os.Signal, 1)
    signal.Notify(sigch, os.Interrupt, syscall.SIGTERM)
    <-sigch
    cancel()  // Trigger graceful shutdown
}()

// Run pool (blocks until context cancelled)
if err := pool.Run(ctx, processFunc); err != nil {
    log.Fatal(err)
}

// Pool automatically:
// 1. Stops accepting new messages
// 2. Waits for inflight messages to complete
// 3. Commits final offsets
// 4. Cleans up resources

Examples

See the examples/ directory for complete examples:

simple/ - Basic usage
Coming soon: with-retry/ - Custom retry logic
Coming soon: with-metrics/ - Prometheus metrics

How It Works

Gap Detection

Burrow tracks which offsets are complete and only commits contiguous offsets:

Consumed:  [0, 1, 2, 3, 4, 5]
Completed: [0, 2, 4, 5]         ← Processing done
Gaps:      [1, 3]               ← Still processing or failed
Committable: 0                  ← Can only commit up to 0

After 1 completes:
Completed: [0, 1, 2, 4, 5]
Committable: 2                  ← Now can commit 0,1,2

After 3 completes:
Completed: [0, 1, 2, 3, 4, 5]
Committable: 5                  ← All done!

This ensures at-least-once delivery: if your app crashes, you restart from the last committed offset and reprocess any gaps.

Performance

Throughput Improvement

Sequential:  20 msgs/sec (50ms per message)
Burrow:      2000 msgs/sec (100 workers)
Improvement: 100x

Actual performance depends on your I/O operations.

Overhead

Memory: O(W) where W = inflight messages (~1KB per message)
Latency: ~1ms coordination overhead per message
Commit: ~5-10ms (synchronous Kafka commit)

Use Cases

Perfect For ✅

API calls to external services
Database queries with high latency
File I/O operations
Image/video processing
Any I/O-bound operation with variable latency

Not Ideal For ❌

CPU-bound operations (use more partitions instead)
Exactly-once semantics required (use Kafka transactions)
Sub-millisecond latency critical (coordination adds overhead)

Architecture

For detailed architecture and design decisions, see DESIGN.md.

Kafka → Poll → Dispatch → [Worker Pool] → Results → Gap Detection → Commit
                              ↓
                         100 goroutines
                         process concurrently

Testing

# Run all tests
go test ./tests/...

# With race detector
go test -race ./tests/...

# Specific test
go test -run TestIntegration_GapDetection ./tests/

Contributing

Contributions welcome! Please:

Fork the repository
Create a feature branch
Add tests for new functionality
Ensure all tests pass with -race
Submit a pull request

License

MIT License - See LICENSE file for details.

Inspiration

Burrow is inspired by:

Eddies - Worker pool pattern
Confluent Parallel Consumer - Gap detection approach
Sift Engineering - At-least-once semantics

Support

Documentation: See DESIGN.md for architecture details
Issues: Report bugs on GitHub Issues
Questions: Open a discussion on GitHub Discussions

Built with ❤️ for the Kafka + Go community

Name		Name	Last commit message	Last commit date
Latest commit History 8 Commits
examples/simple		examples/simple
tests		tests
.gitignore		.gitignore
DESIGN.md		DESIGN.md
README.md		README.md
config.go		config.go
gap.go		gap.go
go.mod		go.mod
go.sum		go.sum
pool.go		pool.go
tracker.go		tracker.go
types.go		types.go

Uh oh!

Folders and files

Latest commit

History

Repository files navigation