NATS Jetstream adds a file storage mode by subject and consumes one message per subject in rotation

[wf-soft]

Proposed change

Add new options

Use case

When consuming stream messages, consumers can take turns receiving messages for each subject, thus ensuring fair consumption

Contribution

No response

[neilalexander]

Converted to a discussion. This is already possible by using the queue subscriptions to have multiple subscribers for a given consumer. See here: https://docs.nats.io/nats-concepts/core-nats/queue#stream-as-a-queue

[wf-soft]

@neilalexander This mode doesn't seem to be suitable for me. If I have thousands of topics that will dynamically increase, how can I subscribe

[wf-soft]

And currently, file storage is not based on themes. How to ensure efficient extraction of theme messages

[wf-soft]

What you said about providing multiple subscribers doesn't seem to be about message rotation, it's just about filtering topics and allocating them in order

[neilalexander]

Queue subscribers can work on the same subject and round-robin. You can have multiple queue subscribers on the same wildcard subscription too for multiple subjects.

I do not know what you mean by "theme messages", in the case of JetStream you would have a stream with one or more subjects based on your configuration and attach consumers to those.

[wf-soft]

@neilalexander The subscription queue group seems to be not suitable for Jetstream. I tried to subscribe but did not receive any messages

var subject = "a.>";
var options = new NatsOpts() { Url = "nats://localhost:4223" };
await using var connection1 = new NatsConnection(options);
var task1 = Task.Run(async () =>
{
    await foreach (var msg in connection1.SubscribeAsync<string>(subject, queueGroup: "My-Workers"))
    {
        Console.WriteLine($"[1][RCV] {msg.Subject}: {msg.Data}\n");
    }
});

await using var connection2 = new NatsConnection(options);
var task2 = Task.Run(async () =>
{
    await foreach (var msg in connection2.SubscribeAsync<string>(subject, queueGroup: "My-Workers"))
    {
        Console.WriteLine($"[2][RCV] {msg.Subject}: {msg.Data}\n");
    }
});
await Task.WhenAll(task1, task2);

[wf-soft]

@neilalexander What I need is the following consumption pattern: first consume all Message 1 messages for each subject, and then consume Message 2 messages for each subject, instead of consuming all Message 1, Message 2, and Message 3 for Subject 1 before consuming Subject 2

[wf-soft]

I am using WorkQueuePolicy

[ripienaar]

We do not have a purely subject based distribution, a busy subject can overwhelm a less busy subject.

There are strategies to group subjects - like you could have message.<bucket>.1 where bucket is calculated and messages will fill the 10 buckets. Thus you can have consumers attached to each of the buckets. But this will still not guarantee fair delivery cross subjects it might just help a bit.

[wf-soft]

@ripienaar I want to extract messages from each subject and allocate them to all consumers when there are multiple subjects. When the subjects with fewer messages are consumed, they will be alternately extracted and allocated to all consumers in the remaining subjects. If there is only one left, all messages will be extracted from the last subject and allocated to all consumers. This fair consumption avoids one subject having a large number of messages while other subjects can only wait for it to be fully completed before being consumed. Do you have any good suggestions

[ripienaar]

We do not have subject based distribution to create the setup you want

[wf-soft]

I thought that the subjects and file shards currently used by Nats should be very compatible with implementing such functional extensions. Alright, thank you for your reply.