JetStream subcribe to Dead Letter Queue

[alberk8]

Is there an .Net example of how get the Dead Letter Queue for Max Deliver ($JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES)?.

[mtmk]

I don't know of a c# example but Jeremy discussed it on YouTube a few weeks back. Have you seen that?

edit: here it is: https://www.youtube.com/watch?v=7Jp3tyCGMZs&t=1510s

[alberk8]

I figured out how to get the DLQ info but cannot figure how to get the payload.

in the CLI with nats stream get EVENTS 1 the payload is shown. What is the equivalent in C#? Thanks.

[mtmk]

you can get directly from the stream. note message data is base64 encoded:

// equivalent of `nats stream get EVENTS 1`

await using var nats = new NatsConnection();
var js = new NatsJSContext(nats);
var stream = await js.GetStreamAsync("EVENTS");

var msg = await stream.GetAsync(new StreamMsgGetRequest { Seq = 1 });

var time = DateTimeOffset.Parse(msg.Message.Time);
var subject = msg.Message.Subject;
var data = Encoding.ASCII.GetString(Convert.FromBase64String(msg.Message.Data));

Console.WriteLine($"received {time} on {subject}:");
Console.WriteLine(data);

[alberk8]

Thank you for the sample code

[alberk8]

I have started a new instance of Docker container. Now I cannot get the DLQ. What am I missing?

 ./nats stream ls
╭──────────────────────────────────────────────────────────────────────────────╮
│                                    Streams                                   │
├────────┬─────────────┬─────────────────────┬──────────┬───────┬──────────────┤
│ Name   │ Description │ Created             │ Messages │ Size  │ Last Message │
├────────┼─────────────┼─────────────────────┼──────────┼───────┼──────────────┤
│ EVENTS │             │ 2024-03-04 12:10:25 │ 1        │ 110 B │ 58.71s       │
│ DLQ    │             │ 2024-03-04 12:10:25 │ 1        │ 287 B │ 35.70s       │
╰────────┴─────────────┴─────────────────────┴──────────┴───────┴──────────────╯

 ./nats stream view DLQ
[1] Subject: $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.EVENTS.processor Received: 2024-03-04T12:11:51+08:00

{"type":"io.nats.jetstream.advisory.v1.max_deliver","id":"fAOhQ2lHPgzu3lS0O8VZTf","timestamp":"2024-03-04T04:11:51.110552679Z","stream":"EVENTS","consumer":"processor","stream_seq":1,"deliveries":5}

 ./nats stream get EVENTS 1
Item: EVENTS#1 received 2024-03-04 04:11:28.102422605 +0000 UTC on Subject data.person

{"Name":"Nats","Address":"2018088782 Random","Serial":"SerialNumber"}

CODE

using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using System.Text;

namespace ConsoleNatsDeadLetterQueue
{
    internal class Program
    {
        static async Task Main(string[] args)
        {
            Console.WriteLine("Hello, World DLQ!");

            var opt = new NatsOpts
            {
                Url = "nats://127.0.0.1:4222",
                Name = "Nats_Connect",

            };

            var consumerName = "processor";
            var ackWait = TimeSpan.FromSeconds(10);
            var ackPolicy = ConsumerConfigAckPolicy.Explicit;
       
            var connect = new NatsConnection(opt);

            var jsContext = new NatsJSContext(connect);

            var dlqstream = await jsContext.GetStreamAsync("EVENTS");

            var stream = await jsContext.CreateStreamAsync(
                new StreamConfig("DLQ", new[] { "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.EVENTS.*" }));

            var consumer = await jsContext.CreateOrUpdateConsumerAsync("DLQ", new ConsumerConfig(consumerName)
            {

                DeliverPolicy = ConsumerConfigDeliverPolicy.All,
                FilterSubject = "data.>",
                AckPolicy = ackPolicy,
                AckWait = ackWait,

            });



            await foreach (NatsJSMsg<byte[]> msg in consumer.ConsumeAsync<byte[]>())
            {
                Console.WriteLine($"DLQ Subject: {msg.Subject} Size: {msg.Size} Subject: {msg.Subject}");

                Console.WriteLine("Data: " + Encoding.UTF8.GetString(msg.Data));

                var md = msg.Metadata;
                if (md.HasValue)
                {

                    Console.WriteLine($"DLQ Num Delivered: {md.Value.NumDelivered} TimeStamp: {md.Value.Timestamp.ToLocalTime()}");
                    Console.WriteLine($"DLQ Consumer Seq: {md.Value.Sequence.Consumer} Stream Seq: {md.Value.Sequence.Stream}");

                    StreamMsgGetRequest streamMsgGetRequest = new StreamMsgGetRequest { Seq = md.Value.Sequence.Stream };
                    var sget = await stream.GetAsync(streamMsgGetRequest);
                    Console.WriteLine("DLQ Message: " + Encoding.UTF8.GetString(Convert.FromBase64String(sget.Message.Data)));

                    var resmsg = await dlqstream.GetAsync(streamMsgGetRequest);
                    Console.WriteLine("DLQ Data: " + Encoding.UTF8.GetString(Convert.FromBase64String(resmsg.Message.Data)));



                }

            }
            Console.WriteLine("Done");
        }
        public class Person
        {
            public string Name { get; set; } = string.Empty;
            public string Address { get; set; } = string.Empty;
            public string Serial { get; set; } = "SerialNumber";
        }
    }
}

[mtmk]

should you use FilterSubject = "data.>", in your consumer?

[mtmk]

ordered consumer uses temporary/ephemeral consumer objects created on the server to track its progress whereas 'normal' consumers create 'durable' consumer objects on the server so that the progress of consuming messages can be tracked across multiple sessions/connections until the consumer is deleted.

Use case wise, ordered consumer is used to read what's in a stream. 'normal' consumer is used to process messages in a stream.

hope it helps :-)

[alberk8]

Thank you for the information. So now I can read and retrieve the message from the DLQ, how do I acknowledge the DLQ has being processed?

[robertmircea]

@mtmk Durable consumer names need to be unique on the server? Let's say I have two consumer apps A and B. Both A and B consume from the same stream, but doing different business processes/logic. Each app has multiple instances to load balance workload. My current understanding is that durable consumer names should be unique per app (e.g all instances of app A should have a a durable consumer name which should be different of durable consumer name of app B). Two or more instances of the same app should share the same durable name. Is that correct?

[mtmk]

yes, that's correct. each consumer is kept on the server identified by their name and each will have its own view of the stream i.e. message offsets, ack/nack status, redeliveries etc. So, different applications should have their own consumers pointing at the same stream and they can be load balanced by running multiple instances of the same application.

[mtmk]

@alberk8 to process the messages from DLQ you need a durable consumer. ordered consumers don't have the notion of ack/nak. If you were having issues reading all the messages using a durable consumer you might have already processed some of the messages, check with nats cli e.g.:

$ nats consumer report EVENTS_STREAM

You might have to delete and create the consumer again (or create another one) to make sure you have can process all messages.