Consumer Group

Overview

A consumer group turns Events Store from a broadcast fan-out into a competing-consumers queue: subscribers sharing the same group split the stored events between them instead of each getting a copy of every event. Reach for this when a durable, ordered event log also needs to scale horizontally — a stream of order updates or audit records where one processor can't keep up, but each event still needs to be handled exactly once by the group as a whole.

It works by passing the same group name to SubscribeToEventsStore on each subscriber alongside a start position such as kubemq.StartFromFirstEvent(). The broker load-balances deliveries across every active member sharing that group and channel; adding another subscriber with the same group name is all it takes to add capacity. Gotchas: the start position belongs to the group's shared read cursor, not to any one subscriber — members joining later pick up wherever the group already is, not from the beginning. Different group names silently mean broadcast instead of load balancing, with no error to warn you. Delivery is exactly-once per group, but a crashed member's in-flight event isn't automatically handed to another member — design processing to be safely restartable.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: events-store/consumer-group
//
// Demonstrates load-balanced event store consumption with consumer groups.
// When multiple subscribers share the same group, each event is delivered
// to exactly one member.
//
// Channel: go-events-store.consumer-group
// Client ID: go-events-store-consumer-group-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

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

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-events-store-consumer-group-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-events-store.consumer-group"
	group := "go-events-store-worker-group"
	received := make(chan struct{})

	// Subscribe with a consumer group for load-balanced event store delivery.
	sub, err := client.SubscribeToEventsStore(ctx, channel, group,
		kubemq.StartFromFirstEvent(),
		kubemq.WithOnEventStoreReceive(func(e *kubemq.EventStoreReceive) {
			fmt.Printf("[ConsumerGroup] seq=%d body=%s\n", e.Sequence, string(e.Body))
			select {
			case received <- struct{}{}:
			default:
			}
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Subscription error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()

	// Send an event.
	_, err = client.SendEventStore(ctx, kubemq.NewEventStore().
		SetChannel(channel).
		SetBody([]byte("group event")).
		SetMetadata("group-demo"))
	if err != nil {
		log.Fatal(err)
	}

	select {
	case <-received:
		fmt.Println("Consumer group demo complete")
	case <-ctx.Done():
		log.Fatal("Timed out")
	}
}

How It Works

1. SubscribeToEventsStore is called with group := "go-events-store-worker-group" and kubemq.StartFromFirstEvent() to replay all stored events and continue with new ones.
2. Consumer groups give each stored event to exactly one member of the group, enabling horizontal scaling of event processors without duplicate processing.
3. The buffered received channel (capacity 1) prevents the callback from blocking the gRPC stream goroutine if the main goroutine hasn't read yet.
4. Scaling out is as simple as adding more subscribers with the same channel and group; the broker load-balances deliveries across all active members.

Related

• Pattern overview
• Go SDK Reference
• Persistent Pub/Sub
• Cancel Subscription