cgnats

cgnats

Strictly ordered message processing for NATS JetStream with Kafka-style consumer groups.

[!WARNING] This project is highly experimental and production use is discouraged for now. The API is subject to breaking changes.

Installation

go get github.com/ramasauskas/cgnats

Features

• Kafka-style consumer groups - Members join groups, receive partition assignments, and coordinate via a central coordinator
• Strict ordering guarantees - Messages within a partition are processed in order with at-least-once delivery
• Automatic rebalancing - Partitions are redistributed when consumers join or leave the group
• Offset management - Durable offset storage using JetStream KV with automatic recovery on restart
• Pluggable partition assignment - Built-in round-robin assigner with support for custom strategies
• High availability - Coordinator leader election with automatic failover
• Lifecycle hooks - Callbacks for rebalancing, partition assignment, commits, and heartbeats for observability

Motivation

NATS JetStream provides powerful streaming capabilities, but achieving total order processing across multiple consumers in a distributed system remains challenging:

• OrderedConsumer guarantees ordering but is limited to a single consumer - no horizontal scaling or high availability
• Solutions like orbit.go/pcgroups help with partitioning but still require MaxAckPending=1 for strict ordering, limiting throughput due to network I/O

cgnats solves this by implementing Kafka-style consumer group coordination on top of JetStream.

This package focuses solely on consumption coordination. It does not handle message publishing or schema design - you are responsible for designing your stream subjects, partitioning strategy, and message format.

Requirements

NATS Server with JetStream enabled

Example

Starting the Coordinator

The coordinator manages consumer group membership and partition assignments.

package main

import (
	"context"
	"log"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/ramasauskas/cgnats"
)

func main() {
	ctx := context.Background()

	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	js, _ := jetstream.New(nc)

	coordinator, err := cgnats.NewCoordinator(ctx, nc, js,
		cgnats.WithCoordinatorSessionTimeout(10*time.Second),
	)
	if err != nil {
		log.Fatal(err)
	}

	// Run blocks until context is cancelled
	coordinator.Run(ctx)
}

[!NOTE] A standalone coordinator binary is on the roadmap, so you won't need to run this in your own process.

Consuming Messages

Consumers join a group and receive partition assignments. The reader handles rebalancing automatically when consumers join or leave the group.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/ramasauskas/cgnats"
)

func main() {
	ctx := context.Background()

	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()

	js, _ := jetstream.New(nc)

	// Map partition numbers to stream and subject
	resolver := func(partition int) (stream, subject string) {
		return "ORDERS", fmt.Sprintf("ORDERS.%d", partition)
	}

	reader, err := cgnats.NewReader(
		nc, js,
		"order-processors", // consumer group name
		4,                  // number of partitions
		cgnats.WithPartitionResolver(resolver),
		cgnats.WithSessionTimeout(10*time.Second),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer reader.Close()

	for {
		msgs, err := reader.Fetch(ctx, 10)
		if err != nil {
			log.Printf("fetch error: %v", err)
			continue
		}

		for _, msg := range msgs {
			log.Printf("partition=%d seq=%d data=%s",
				msg.Partition, msg.Sequence, msg.Data)
		}

		if err := reader.Ack(ctx, msgs...); err != nil {
			log.Printf("ack error: %v", err)
		}
	}
}

Roadmap

• Standalone coordinator binary (currently you need to run the coordinator in your own process)
• More chaotic E2E tests to ensure consistency

Acknowledgements

The reader API is inspired by segmentio/kafka-go.