Competing Consumer Pattern

[thenhawke]

I'm not sure the way that I'm creating a durable push consumer is correct, I'm trying to configure a competing consumer pattern on a stream.

This is how I create the stream:

_jetStreamManagement.AddStream(StreamConfiguration.Builder()
    .WithName("NewOrders")
    .WithSubjects("Orders.Created.*")
    .WithRetentionPolicy(RetentionPolicy.WorkQueue)
    .Build());

This is how I create the consumer:

_jetStreamManagement.AddOrUpdateConsumer("NewOrders", ConsumerConfiguration.Builder()
    .WithDurable("NewOrderProcessor")
    .WithDeliverSubject("NewOrdersToProcess")
    .WithDeliverGroup("NewOrderProcessors")
    .WithAckPolicy(AckPolicy.Explicit)
    .Build());

I create a subscriber like this:

_subscription = _jetStream.PushSubscribeAsync("Orders.Created.*", HandleOrderCreated, false, PushSubscribeOptions.Builder()
    .WithStream("NewOrders")
    .WithDurable("NewOrderProcessor")
    .WithDeliverSubject("NewOrdersToProcess")
    .WithDeliverGroup("NewOrderProcessors")
    .Build());

The problem with this is when I connect the above subscriber, the subscriber doesn't receive any messages. If I add another subscription with nats con sub NewOrders NewOrderProcessor --deliver-group=NewOrderProcessors then the queued messages drain into the command line subscription and get printed and every message is then also then delivered to the subscription from the above code. If I create a second subscription from the command line then the new messages are sent round robin to the two command line subscribers, while every single message is also sent to the .NET subscription.

It appears that the nats consumer sub NewOrders NewOrderProcessor --deliver-group=NewOrderProcessors command works exactly how I want with each message being delivered to only one subscriber in a round robin fashion and being removed from the queue when acked, I just can't seem to figure out how to replicate the behavior with the .NET library correctly.

[scottf]

Your current subscribe is making a conflicting consumer of the same name that does not match the consumer you have already created.
Change your subscribe to "bind" to your existing consumer.

_subscription = _jetStream.PushSubscribeAsync(null, HandleOrderCreated, false, 
    PushSubscribeOptions.BindTo("NewOrders", "NewOrderProcessor"));

[thenhawke]

Thanks! That is what I was looking for, now I see that example in the docs. The BindTo as described is throwing a SUB-90014 - Existing consumer is configured as a queue / deliver group. I'm creating the stream and consumer as above and I want the deliver group, so I'm not sure how to bind that. I tried adding the deliver group via the long hand configuration which doesn't throw, but it has the same behavior described above where it doesn't get any messages until another subscription is created on the command line, then receives everything.

// This throws the 90014
_subscription = _jetStream.PushSubscribeAsync(null, HandleOrderCreated, false, 
    PushSubscribeOptions.BindTo("NewOrders", "NewOrderProcessor"));

// This configuration doesn't throw, but it has the same behavior described above, 
// it receives no messages until another subscription is added via command line, then gets everything in the stream
var options = PushSubscribeOptions.Builder()
    .WithStream("NewOrders")
    .WithDurable("NewOrderProcessor")
    .WithDeliverGroup("NewOrderProcessors")
    .WithBind(true)
    .Build();
_subscription = _jetStream.PushSubscribeAsync(null, HandleOrderCreated, false, options);

Here is the consumer configuration:

Information for Consumer NewOrders > NewOrderProcessor created 2023-06-01T11:03:08-04:00

Configuration:

        Durable Name: NewOrderProcessor
    Delivery Subject: NewOrdersToProcess
      Deliver Policy: All
 Deliver Queue Group: NewOrderProcessors
          Ack Policy: Explicit
            Ack Wait: 30s
       Replay Policy: Instant
     Max Ack Pending: 1,000
        Flow Control: false

How do I bind to this consumer correctly?

[scottf]

Sorry I missed putting the queue name in the subscribe call. Here is a full example.

using (IConnection c = new ConnectionFactory().CreateConnection(opts))
{
    IJetStreamManagement _jetStreamManagement = c.CreateJetStreamManagementContext();
    IJetStream _jetStream = c.CreateJetStreamContext();
            
    _jetStreamManagement.AddStream(StreamConfiguration.Builder()
        .WithName("NewOrders")
        .WithSubjects("Orders.Created.*")
        .WithRetentionPolicy(RetentionPolicy.WorkQueue)
        .WithStorageType(StorageType.Memory) // for testing
        .Build());

    // for testing, always gives a clean set of messages in case I re-run the example
    _jetStreamManagement.PurgeStream("NewOrders"); 
    
    for (int x = 0; x < 20; x++)
    {
        _jetStream.Publish($"Orders.Created.{x}", Encoding.UTF8.GetBytes($"Order For {x}"));
    }
    
    ConsumerConfiguration cc = ConsumerConfiguration.Builder()
        .WithDurable("NewOrderProcessor")
        .WithDeliverSubject("NewOrdersToProcess")
        .WithDeliverGroup("NewOrderProcessors")
        .WithAckPolicy(AckPolicy.Explicit) // explicit is default, but it documents the code this way!
        .Build();
    
    _jetStreamManagement.AddOrUpdateConsumer("NewOrders", cc);
    
    CountdownEvent latch = new CountdownEvent(20);
    EventHandler<MsgHandlerEventArgs> HandleOrderCreated = (sender, eventArgs) =>
    {
        eventArgs.Message.Ack();
        Console.WriteLine(Encoding.UTF8.GetString(eventArgs.Message.Data));
        latch.Signal();
    };

    PushSubscribeOptions pso = PushSubscribeOptions.BindTo("NewOrders", "NewOrderProcessor");
    _jetStream.PushSubscribeAsync(null, "NewOrderProcessors", HandleOrderCreated, false, pso);

    latch.Wait(10_000);
}

[thenhawke]

Awesome, thanks! Binding to the existing consumer and passing the queue name has everything working as expected.