What Happens When multiple Consumers share a same durable Name with different topic Filter in WorkQueue Jetstream?

[YashasAnand]

I have a stream in WorkQueue based Retenetion created with Jetstream.

I have instances of consumer application running. I have a consumer which starts consuming when app starts & its Durable Name is same (ex: OrderConsumer) .

Now in applicatin 1, if i create OrderConsumer#0 with SubjectFilter as orders.app1
& in application 2, if i create OrderConsumer#0 with SubjectFilter as orders.app2

I observe that if i produce message to orders.app1, The app 2 which has intrest on orders.app2 is actually getting orders.app1 message.
Why is it getting loadbalanced & not consuming by correct app with correct filter?

Can someone pls explain to me this concept of what is happening here? Ideally my understanding was only app1 would receive this message.

[Jarema]

You cannot have two consumers with the same name. Server will not allow it. I suspect you have two subscriptions to the same consumer which explains what you see.

Can you share a code snippet, stream info and consumer info?

[YashasAnand]

Thank you so much for your quick response @Jarema , Pls find my code below

Stream Creation:

static async Task Main(string[] args)
{
await using var nats = new NatsConnection();
var js = new NatsJSContext(nats);
await nats.ConnectAsync();
var streamConfig = new StreamConfig(
name: "orders", // Stream name
subjects: new[] { "orders.>" } // Stream subjects
)
{
Discard = StreamConfigDiscard.Old, // Discard old messages when limits are reached
MaxBytes = 10L * 1024 * 1024 ,
Retention = StreamConfigRetention.Workqueue // Set the maximum size of the stream to 10MB
};
await js.CreateStreamAsync(streamConfig);
}

Publisher which will just send to orders.app1 or orders.app2:
var ack = await js.PublishAsync(subject: "orders.app2", data: "order 1");
ack.EnsureSuccess();

Consumer App1:

static async Task Main(string[] args)
{
await using var nats = new NatsConnection();
var js = new NatsJSContext(nats);
await nats.ConnectAsync();
var ConsumerConfig = new ConsumerConfig
{
Name = "OrderConsumer",
DurableName = "OrderConsumer",
AckPolicy = ConsumerConfigAckPolicy.Explicit,
FilterSubject = "orders.app1"
};
var consumer = await js.CreateOrUpdateConsumerAsync("orders", ConsumerConfig);
await foreach (var jsMsg in consumer.ConsumeAsync())
{
Console.WriteLine($"Processed: {jsMsg.Data}");
await jsMsg.AckAsync();
}
}

Consumer App 2: 

class Program
{
static async Task Main(string[] args)
{
await using var nats = new NatsConnection();
var js = new NatsJSContext(nats);
await nats.ConnectAsync();
var ConsumerConfig = new ConsumerConfig
{
Name = "OrderConsumer",
DurableName = "OrderConsumer",
AckPolicy = ConsumerConfigAckPolicy.Explicit,
FilterSubject = "orders.app2"
};
var consumer = await js.CreateOrUpdateConsumerAsync("orders", ConsumerConfig);
await foreach (var jsMsg in consumer.ConsumeAsync())
{
Console.WriteLine($"Processed: {jsMsg.Data}");
await jsMsg.AckAsync();
}
}
}

I understand 2 consumers cannot use same name, but these are 2 different consumer apps listening to same consumer with different filterSubject.

Requesting you to pls help me understand what is wrong.

[Jarema]

Filter subject is part of consumer config, not subscription to it. You cannot subscribe with different subject to the same consumer.

Please check the Consumer Info to confirm.

I suspect that whichever app starts later, updates the consumer filter, and both apps will get the updates filtered messages.

[YashasAnand]

Thank you for claifying @Jarema.