Synadia.Orbit.PCGroups 1.0.0-preview.9

Synadia.Orbit.PCGroups

Partitioned Consumer Groups (PCGroups) for NATS JetStream. This library enables horizontal scaling of message processing by automatically distributing partitions across consumer group members.

Requirements: NATS Server 2.11+ (for Priority Groups/Pinning support)

Features

• Static Consumer Groups: Fixed membership defined at creation time
• Elastic Consumer Groups: Dynamic membership changes at runtime
• Automatic Partition Distribution: Partitions are automatically balanced across members
• Custom Partition Mappings: Explicit control over which partitions each member handles
• Self-Healing: Consumers automatically recover from failures
• KV-based Configuration: Group configurations stored in NATS KV for coordination

Installation

dotnet add package Synadia.Orbit.PCGroups

Quick Start

Static Consumer Groups

Static groups have a fixed membership that cannot change after creation. The stream must be configured with a subject transform to add partition prefixes.

using NATS.Client.JetStream.Models;
using NATS.Net;
using Synadia.Orbit.PCGroups.Static;

await using var nats = new NatsClient();
var js = nats.CreateJetStreamContext();

// Create the stream with subject transform for partitioning
// The transform adds a partition prefix based on the wildcard token
await js.CreateStreamAsync(new StreamConfig("orders", ["orders.*"])
{
    SubjectTransform = new SubjectTransform
    {
        Src = "orders.*",
        Dest = "{{partition(3,1)}}.orders.{{wildcard(1)}}",
    },
});

// Create a static consumer group with 3 partitions
await js.CreatePcgStaticAsync(
    streamName: "orders",
    consumerGroupName: "order-processors",
    maxNumMembers: 3,
    filters: ["orders.*"]);

// Publish messages - they get transformed to {partition}.orders.{id}
await js.PublishAsync("orders.123", new Order("ORD-123", "CUST-1", 99.99m));

// Start consuming using async enumerable
await foreach (var msg in js.ConsumePcgStaticAsync<Order>(
    streamName: "orders",
    consumerGroupName: "order-processors",
    memberName: "worker-1"))
{
    Console.WriteLine($"Processing order: {msg.Subject} - {msg.Data}");
    await msg.AckAsync();
}

record Order(string OrderId, string CustomerId, decimal Amount);

Elastic Consumer Groups

Elastic groups allow dynamic membership changes at runtime. The library automatically creates a work-queue stream with the appropriate transforms.

using System.Threading.Channels;
using NATS.Client.JetStream.Models;
using NATS.Net;
using Synadia.Orbit.PCGroups;
using Synadia.Orbit.PCGroups.Elastic;

await using var nats = new NatsClient();
var js = nats.CreateJetStreamContext();

// Create the source stream (no transform needed - elastic creates work-queue stream)
await js.CreateStreamAsync(new StreamConfig("events", ["events.*"]));

// Create an elastic consumer group
// Partitioning is based on the first wildcard token in the subject
await js.CreatePcgElasticAsync(
    streamName: "events",
    consumerGroupName: "event-processors",
    maxNumMembers: 10,
    partitioningFilters: [new NatsPcgPartitioningFilter("events.*", [1])]);

// Add members dynamically - partitions will be distributed across them
string[] members = ["worker-1", "worker-2", "worker-3"];
await js.AddPcgElasticMembersAsync("events", "event-processors", members);

// Publish some messages
await js.PublishAsync("events.user123", new Event("EVT-1", "user123", "click"));

// Use a channel to aggregate messages from multiple workers
var channel = Channel.CreateUnbounded<(string Worker, NatsPcgMsg<Event> Msg)>();
using var cts = new CancellationTokenSource();

// Start a consumer task for each worker
var consumerTasks = members.Select(worker => Task.Run(async () =>
{
    try
    {
        await foreach (var msg in js.ConsumePcgElasticAsync<Event>(
            streamName: "events",
            consumerGroupName: "event-processors",
            memberName: worker,
            cancellationToken: cts.Token))
        {
            await channel.Writer.WriteAsync((worker, msg), cts.Token);
        }
    }
    catch (OperationCanceledException)
    {
        // Expected when cancelled
    }
})).ToArray();

// Process messages from all workers
await foreach (var (worker, msg) in channel.Reader.ReadAllAsync(cts.Token))
{
    Console.WriteLine($"[{worker}] Processing event: {msg.Subject} - {msg.Data}");
    await msg.AckAsync();
}

record Event(string EventId, string UserId, string Type);

Static vs Elastic Comparison

Graceful Shutdown

PCGroups supports NATS .NET drain-on-dispose. Enable it on the connection to let in-flight JetStream consumer messages already buffered by the client continue to flow through the PCGroups async enumerable before the connection is closed:

await using var nats = new NatsConnection(new NatsOpts
{
    DrainSubscriptionsOnDispose = true,
    ConsumerDrainOnDisposeTimeout = TimeSpan.FromSeconds(30),
});

Without DrainSubscriptionsOnDispose, NATS .NET closes the socket first during connection disposal and buffered consumer messages may remain pending until AckWait expires.

To drain without disposing the connection, pass drainOnCancel: true to the consume call. A graceful stop then stops pulling, delivers the messages already buffered by the client so handlers can ACK them on the still open connection, and completes the enumerable instead of stopping immediately. A graceful stop is either cancelling the supplied token or an internal lifecycle event such as a membership change; buffered messages are valid for this consumer either way:

using var cts = new CancellationTokenSource();

await foreach (var msg in js.ConsumePcgStaticAsync<string>(
    streamName, groupName, memberName, drainOnCancel: true, cancellationToken: cts.Token))
{
    // cts.Cancel() lets the buffered messages drain through this loop before it ends.
    await msg.AckAsync();
}

Custom Partition Mappings

For fine-grained control over partition distribution:

using NATS.Client.JetStream.Models;
using NATS.Net;
using Synadia.Orbit.PCGroups;
using Synadia.Orbit.PCGroups.Static;
using Synadia.Orbit.PCGroups.Elastic;

await using var nats = new NatsClient();
var js = nats.CreateJetStreamContext();

// Define explicit member-to-partition mappings
var mappings = new[]
{
    new NatsPcgMemberMapping("high-priority-worker", [0, 1, 2]),
    new NatsPcgMemberMapping("low-priority-worker", [3, 4, 5]),
};

// For static groups - stream needs subject transform
await js.CreateStreamAsync(new StreamConfig("orders", ["orders.*"])
{
    SubjectTransform = new SubjectTransform
    {
        Src = "orders.*",
        Dest = "{{partition(6,1)}}.orders.{{wildcard(1)}}",
    },
});

await js.CreatePcgStaticAsync("orders", "processors", maxNumMembers: 6,
    filters: ["orders.*"], memberMappings: mappings);

// For elastic groups - no transform needed on source stream
await js.CreateStreamAsync(new StreamConfig("events", ["events.*"]));

await js.CreatePcgElasticAsync("events", "processors", maxNumMembers: 6,
    partitioningFilters: [new NatsPcgPartitioningFilter("events.*", [1])]);
await js.SetPcgElasticMemberMappingsAsync("events", "processors", mappings);

Subject Transform Syntax

For static consumer groups, the stream must use subject transforms to add partition prefixes:

• {{partition(N,wildcards)}} - Computes partition (0 to N-1) based on specified wildcard positions
• {{wildcard(N)}} - References the Nth wildcard token from the source subject (1-indexed)

Example: For orders.* with transform {{partition(3,1)}}.orders.{{wildcard(1)}}:

• orders.ABC becomes 0.orders.ABC, 1.orders.ABC, or 2.orders.ABC (based on hash of "ABC")

API Reference

Static Consumer Groups (Extension methods on INatsJSContext)

• CreatePcgStaticAsync - Create a new static consumer group
• GetPcgStaticConfigAsync - Get configuration for an existing group
• ConsumePcgStaticAsync - Start consuming messages (returns IAsyncEnumerable<NatsPcgMsg<T>>)
• DeletePcgStaticAsync - Delete a consumer group
• ListPcgStaticAsync - List all consumer groups for a stream
• ListPcgStaticActiveMembersAsync - List active members
• PcgStaticMemberStepDownAsync - Force a member to step down

Elastic Consumer Groups (Extension methods on INatsJSContext)

• CreatePcgElasticAsync - Create a new elastic consumer group
• GetPcgElasticConfigAsync - Get configuration for an existing group
• ConsumePcgElasticAsync - Start consuming messages (returns IAsyncEnumerable<NatsPcgMsg<T>>)
• DeletePcgElasticAsync - Delete a consumer group and its work-queue stream
• ListPcgElasticAsync - List all consumer groups for a stream
• ListPcgElasticActiveMembersAsync - List active members
• AddPcgElasticMembersAsync - Add members to the group
• DeletePcgElasticMembersAsync - Remove members from the group
• SetPcgElasticMemberMappingsAsync - Set explicit partition mappings
• DeletePcgElasticMemberMappingsAsync - Remove mappings (revert to auto-distribution)
• IsInPcgElasticMembershipAndActiveAsync - Check if a member is in the group and active
• GetPcgElasticPartitionFilters - Get partition filters for a member (extension on NatsPcgElasticConfig)
• PcgElasticMemberStepDownAsync - Force a member to step down

Validation (Static class NatsPcgMemberMappingValidator)

• Validate - Validate member mappings (checks for duplicates, overlaps, out-of-range partitions)
• ValidateFilterAndWildcards - Validate filter and partitioning wildcards for elastic groups

How It Works

1. Partitioning: Messages are assigned to partitions (0 to maxMembers-1) based on subject content
2. Distribution: Partitions are distributed across active members
3. Pinning: Each member "pins" to its assigned partitions using priority groups
4. Coordination: Configuration stored in NATS KV enables coordination
5. Self-healing: Members watch for configuration changes and automatically adjust

License

Apache License 2.0