Why does Consumer 1 eats Consumer 2's messages?

[Exagram]

I have a bunch of microservices that publish and subscribe to data events (implemented as NATS messages).

I chose Jetstream and pull-based consumers because they provide persistence, message deduplication and sequence ordering which will be useful for exactly-once delivery as well as providing some fault tolerance for crashing of microservices.

Initially I thought I could configure my consumers can handle a list of selective subjects i.e. STREAM.catalog.*.success, STREAM.catalog.*.fail but I don't see such a feature. What I do see is a wildcard feature i.e. STREAM.> which unfortunately will retrieve all messages (:doh:) but at least it will offer me the messages I want to subscribe to.

Thus I send out all messages to all my microservices:

• I have a NATS stream calle STREAM with filter "STREAM.>" (Everything)
• I have two NATS consumers C1 and C2 with filters "STREAM.>" (Everything)

Both C1 and C2 are has pull-based consumers:

var context = connection.CreateJetStreamContext();
var options = PullSubscribeOptions.Builder()
    .WithStream("STREAM")
    .WithDurable("C1") // The other one has C2
    .Build();
var subscription = context.PullSubscribe("STREAM.>", options);
var msgs = subscription.Fetch(1, 1000);
if (!msgs.Any()) { ... }
...
msg.Ack();

Unfortunately when the C1 microservice publishes STREAM.catalog.insert.success it receives the same message while C2 which listens to the exact same stream/filter doesn't receive it.

Why is this the case? The two consumers have differing durable names. Even though they consume the same filters shouldn't they receive the same messages?

Is this because C1 ack's the message? If so I changed C1 so that it nack's the message but still no good:

msg.Nak();

Am I missing something here?

[scottf]

@Exagram Can you please show the stream configuation? The only thing I can think of is that the stream is configured as RetentionPolicy.WorkQueue, otherwise both consumers should get the message.
https://docs.nats.io/using-nats/developer/develop_jetstream/model_deep_dive#stream-limits-retention-and-policy

[Exagram]

Hi scottf,

It is a File - Limits stream which should be hold messages for up to a year.

PS C:\Users\USERNAME\lib\nats-0.0.28-windows-amd64> .\nats.exe str info STREAM
Information for Stream STREAM created 2022-02-04T19:57:29-05:00

Configuration:

             Subjects: STREAM.>
     Acknowledgements: true
            Retention: File - Limits
             Replicas: 1
       Discard Policy: Old
     Duplicate Window: 2m0s
    Allows Msg Delete: true
         Allows Purge: true
       Allows Rollups: false
     Maximum Messages: unlimited
        Maximum Bytes: unlimited
          Maximum Age: 1y0d0h0m0s
 Maximum Message Size: unlimited
    Maximum Consumers: unlimited


State:

             Messages: 101
                Bytes: 81 KiB
             FirstSeq: 139 @ 2022-02-26T04:13:32 UTC
              LastSeq: 239 @ 2022-03-02T00:08:23 UTC
     Active Consumers: 4

I will post source code for a minimal replicable solution in .NET 6 in a bit...

[Exagram]

Minimal Verifiable PoC

In VS create a .NET 6 console app:

Program.cs:

// See https://aka.ms/new-console-template for more information
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

await Host.CreateDefaultBuilder(args)
    .ConfigureServices((c, services) =>
    {
        services.AddHostedService<NatsService>();
    })
    .RunConsoleAsync();

NatsService.cs:

// See https://aka.ms/new-console-template for more information
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using NATS.Client;
using NATS.Client.JetStream;
using System.Text;

internal class NatsService : BackgroundService
{
    private readonly ILogger<NatsService> logger;
    private readonly IConfiguration configuration;

    public NatsService(ILogger<NatsService> logger, IConfiguration configuration)
    {
        this.logger = logger;
        this.configuration = configuration;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var options = ConnectionFactory.GetDefaultOptions();
        options.Url = "nats://localhost:4222";
        options.Name = "test";
        options.Timeout = 3000;
        options.Verbose = true;

        options.AsyncErrorEventHandler += (sender, args) =>
        {
            logger.LogError($"NATS: Error: Server: {args.Conn.ConnectedUrl}. Message: {args.Error}. Subject: {args.Subscription.Subject}");
        };

        options.ServerDiscoveredEventHandler += (sender, args) =>
        {
            logger.LogInformation($"NATS: A new server has joined the cluster: {string.Join(", ", args.Conn.DiscoveredServers)}");
        };

        options.ClosedEventHandler += (sender, args) =>
        {
            logger.LogInformation($"NATS: Connection Closed: {args.Error?.Message}. Server: {args.Conn.ConnectedUrl}");
        };

        options.DisconnectedEventHandler += (sender, args) =>
        {
            logger.LogInformation($"NATS: Connection Disconnected: {args.Error?.Message}. Server: {args.Conn.ConnectedUrl}.");
        };
        var c = new ConnectionFactory().CreateConnection(options);
        if (!string.IsNullOrEmpty(configuration["PublishSubject"]))
        {
            c.Publish(new Msg(configuration["PublishSubject"], Encoding.UTF8.GetBytes("Hello world")));
        }
        while (!stoppingToken.IsCancellationRequested)
        {
            logger.LogInformation("Pulling messages from NATS");
            var context = c.CreateJetStreamContext();
            var pullOptions = PullSubscribeOptions.Builder()
                .WithStream(configuration["Stream"])
                .WithDurable(configuration["Durable"])
                .Build();
            var subscription = context.PullSubscribe(configuration["Subject"], pullOptions);
            var msgs = subscription.Fetch(1, 1000);
            foreach (var msg in msgs)
            {
                logger.LogInformation($"Msg {msg.Subject} {Encoding.UTF8.GetString(msg.Data)}");
                msg.Ack();
            }
            await Task.Delay(1000);
        }
    }
}

In Powershell run:

• Terminal 1: $env:Stream = "STREAM"; $env:Durable = "CONSUMER2"; $env:Subject = "STREAM.>"; .\NatsNetPoC.exe
• Terminal 2: $env:Stream = "STREAM"; $env:Durable = "CONSUMER1"; $env:Subject = "STREAM.>"; $env:PublishSubject = "STREAM.catalog.add.success"; .\NatsNetPoC.exe

Discussion

• I wasn't able to replicate my bug. Both pull-based consumers were able to consume the same message as expected. I'll leave this for posterity if anybody has the same problem. Whatever issue I'm facing must be somewhere else.
• What is good messaging design? If all my microservices listen to "STREAM.>" that will cause a lot of useless traffic.
• Was my retention choice Files - Limits valid for my use case: multiples services i.e. Auditing Service and Notifications Service should consume a message i.e. "STREAM.orders.success"?
• Would an alternative retention choice Worker Queue be a better solution for the above scenario? For this case I would send messages "STREAM.audit.order" and "STREAM.notify.order" instead of "STREAM.orders.success"
• I'm getting decision paralysis here...

[scottf]

1. You can share the the workload between consumers by adding them to a deliver group aka queue. Consumer configuration has this option and it should be specified during subscription creation. Yes we are currently planning on making the api easier next major version.
2. The way we handle subjects is also being revamped, allowing for partitioning and possibly better filtering. In the meantime, my suggestion is to split your subjects into domains. So while the stream configuration subject can be STREAM.> you would likely want to split your consumers by filter. You can use STREAM.orders.> , STREAM.notify.> etc. You can also do thing like STREAM.*.fail and that consumer would handle all types of failures. This doc is a good read. https://docs.nats.io/nats-concepts/subjects
3. There is an excellent description of Retention Policy here: https://docs.nats.io/nats-concepts/jetstream#retention-policies-and-limits Limits is the default. workqueue policy removes the message from the stream once it is consumed/acked, preventing it from being re-read. This may or may not be the behavior you want.