watermill-kafka-franz

README

Watermill Kafka Franz

A Watermill Pub/Sub implementation for Apache Kafka using the modern franz-go client library.

Features

• Full Watermill Pub/Sub interface compliance
• Consumer groups with automatic partition rebalancing
• At-least-once delivery semantics
• Context-based cancellation and graceful shutdown
• Configurable marshaling strategies
• SASL authentication and TLS support
• Idiomatic Go with functional configuration
• High performance (lock-free internals)

Installation

go get github.com/dubyte/watermill-kafka-franz

Quick Start

Publisher

package main

import (
    "context"
    "log"
    
    "github.com/dubyte/watermill-kafka-franz/pkg/kafka"
    "github.com/ThreeDotsLabs/watermill/message"
)

func main() {
    publisher, err := kafka.NewPublisher(kafka.PublisherConfig{
        Brokers: []string{"localhost:9092"},
    }, nil)
    if err != nil {
        log.Fatal(err)
    }
    defer publisher.Close()
    
    msg := message.NewMessage("id-1", []byte("hello world"))
    err = publisher.Publish("my-topic", msg)
    if err != nil {
        log.Fatal(err)
    }
}

Subscriber

subscriber, err := kafka.NewSubscriber(kafka.SubscriberConfig{
    Brokers:       []string{"localhost:9092"},
    ConsumerGroup: "my-consumer-group",
}, nil)
if err != nil {
    log.Fatal(err)
}
defer subscriber.Close()

messages, err := subscriber.Subscribe(context.Background(), "my-topic")
if err != nil {
    log.Fatal(err)
}

for msg := range messages {
    log.Printf("Received: %s", msg.Payload)
    msg.Ack()
}

Configuration

Publisher Options

config := kafka.PublisherConfig{
    Brokers:               []string{"localhost:9092"},
    Marshaler:             kafka.DefaultMarshaler{},
    MaxBufferedRecords:    10000,
    ProduceRequestTimeout: 10 * time.Second,
    BatchMaxBytes:         1 << 20, // 1MB
    Compression:           []kgo.CompressionCodec{kgo.SnappyCompression()},
    ClientID:              "my-app",
}

Subscriber Options

config := kafka.SubscriberConfig{
    Brokers:                []string{"localhost:9092"},
    Unmarshaler:            kafka.DefaultMarshaler{},
    ConsumerGroup:          "my-group",
    AutoOffsetReset:        "earliest", // or "latest", "none"
    HeartbeatInterval:      3 * time.Second,
    SessionTimeout:         45 * time.Second,
    AutoCommitInterval:     5 * time.Second,
    DisableAutoCommit:      false,           // set true for manual commits (e.g. exactly-once processors)
    CommitTimeout:          10 * time.Second, // timeout for manual commits when DisableAutoCommit=true
    NackResendSleep:        100 * time.Millisecond,
    FetchMaxBytes:          50 << 20, // 50MB
    ClientID:               "my-app",
    // OnUnmarshalError: nil (default) = fail fast on poison pills
    // OnUnmarshalError: kafka.SkipUnmarshalErrorHandler(logger) = skip and continue
}

Advanced Usage

Custom Marshaler

type MyMarshaler struct{}

func (m MyMarshaler) Marshal(topic string, msg *message.Message) (*kgo.Record, error) {
    // Custom serialization logic
}

func (m MyMarshaler) Unmarshal(record *kgo.Record) (*message.Message, error) {
    // Custom deserialization logic
}

Auto-Creating Topics

The subscriber implements Watermill's SubscribeInitializer interface. Call SubscribeInitialize to create the topic before consuming (idempotent — a no-op if the topic already exists):

err := subscriber.SubscribeInitialize("my-topic")
if err != nil {
    log.Fatal(err)
}

Partition count and replication factor are controlled by InitializeTopicPartitions and InitializeTopicReplicationFactor on SubscriberConfig (both default to 1).

Poison Pill Handling (Unmarshal Errors)

By default, if a Kafka record cannot be unmarshalled, the subscriber fails fast: it logs the error and stops the goroutine without committing the offset. The record is redelivered on the next startup. This is the safe default for financial systems.

BREAKING CHANGE (v0.x → current): Previous versions silently committed and skipped unprocessable records. To restore that behavior, set OnUnmarshalError to SkipUnmarshalErrorHandler.

Configure via SubscriberConfig.OnUnmarshalError:

// Default (nil): fail fast — subscriber stops, offset not committed, record redelivered.
config := kafka.DefaultSubscriberConfig()
// config.OnUnmarshalError is nil → fail-fast

// Skip and continue (previous behavior):
config.OnUnmarshalError = kafka.SkipUnmarshalErrorHandler(logger)

// Custom handler (e.g. send to DLQ):
config.OnUnmarshalError = func(ctx context.Context, record *kgo.Record, err error) error {
    // publish record.Value to a dead-letter topic, then return nil to skip
    return dlqPublisher.Publish(ctx, record, err)
    // or return err to stop the subscriber
}

OnUnmarshalError return value	Behavior
Field is nil (default)	Fail fast: log error, stop goroutine, no commit — record redelivered
Callback returns nil	Skip: commit offset, continue to next record
Callback returns non-nil error	Stop: log error, stop goroutine, no commit

Context Metadata

Access Kafka metadata from message context:

partition, ok := kafka.PartitionFromContext(msg.Context())
offset, ok := kafka.OffsetFromContext(msg.Context())
timestamp, ok := kafka.MessageTimestampFromContext(msg.Context())
key, ok := kafka.MessageKeyFromContext(msg.Context())

You can also set context values when building custom middleware or marshalers:

ctx = kafka.ContextWithPartition(ctx, partition)
ctx = kafka.ContextWithOffset(ctx, offset)
ctx = kafka.ContextWithTimestamp(ctx, timestamp)
ctx = kafka.ContextWithKey(ctx, key)

Authentication

SASL/PLAIN

config.SASLMechanism = plain.Auth{
    User: "username",
    Pass: "password",
}.AsMechanism()

SASL/SCRAM

config.SASLMechanism = scram.Auth{
    User: "username",
    Pass: "password",
}.AsSha256Mechanism()

TLS

config.TLS = &tls.Config{
    // Your TLS configuration
}

Testing

Tests are organised in three layers. See TESTING.md for the full methodology.

Layer	What	Location	Needs broker
Unit	Config, marshaling, state machines	pkg/kafka/*_test.go	No
Watermill compliance	message.Publisher / message.Subscriber contract via watermill's test suite	pkg/kafka/*_integration_test.go	Yes
Kafka behaviour	At-least-once delivery, network faults, rebalancing, poison pills	tests/integration/	Yes + Toxiproxy

Any file that requires a broker carries //go:build integration.
Running go test ./... without that tag is always safe — no broker, no hangs.

Quick start

# Unit tests — no broker required
make test-short

# All tests — starts Redpanda + Toxiproxy, runs, tears down
make test

# Integration tests against an already-running stack
make docker-up && make wait-for-redpanda
make test-integration

The stack uses Redpanda (Kafka-compatible, ~2 s startup) and Toxiproxy for network-fault injection.

Why Franz-Go?

This library uses franz-go instead of Sarama because:

• Modern Go: Uses latest Go features, cleaner API
• Unified Client: Single client for producing and consuming
• Performance: Lock-free internals, better throughput
• Context Support: Native context.Context support
• Error Handling: Better error types with kerr package
• Active Development: Well-maintained with frequent updates

Comparison with Watermill-Kafka

Feature	Watermill-Kafka (Sarama)	Watermill-Kafka-Franz
API Style	Struct-based config	Functional options + Struct
Client	Separate producer/consumer	Unified client
Context	Limited	Native support
Compression	Limited	Multiple codecs
Performance	Good	Excellent

Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure all tests pass
5. Submit a pull request

License

MIT License - see LICENSE file for details.