[Help needed with Java client] - Jetstream pub/sub - queue not considered, messages broadcast to the consumers in same queue group

[rpuppo]

Hello,

I'm passing hard times trying to make multiple consumers in one single queue group sharing the load on a subject, belonging to a jetstream stream.

Here is the api call I'm using:
JetStreamSubscription subscribe(String subject, String queue, Dispatcher dispatcher, MessageHandler handler, boolean autoAck, PushSubscribeOptions options) throws IOException, JetStreamApiException;

The aim is to call this multiple time, per consumer thread, always specifying same subject, same queue and same PushSubscribeOptions and get the queue well load balanced across my consumers, but what I got is just a broad cast per consumer instance.
Dispatcher is used to make async consumption, per consumer instance.

Here below my code for a single consumer(launched on multiple instances, during test) :
Dispatcher natsJetstreamDispatcher;
Connection nc;
JetStreamManagement jsmConsumer;

    log.info("createConsumer on topic {} with subscription name {}", topic, subscriptionName);
    try {
        final Options connOptions = new Options.Builder().server(config.natsHostUrl).maxReconnects(5).build();
        nc = Nats.connect(connOptions);
        jsmConsumer = nc.jetStreamManagement();
        final JetStream js = nc.jetStream();
        final PushSubscribeOptions consOptions = PushSubscribeOptions.builder()
                .stream(config.streamName)
                .deliverSubject("bench")
                .deliverGroup(subscriptionName)
                .build();

        final MessageHandler msgHandler = (msg -> consumerCallback.messageReceived(msg.getData(),
                Long.parseLong(msg.getHeaders().get(NatsJetstreamBenchmarkProducer.TIMESTAMP_HEADER).get(0))));

        natsJetstreamDispatcher = nc.createDispatcher();
        js.subscribe(topic, subscriptionName, natsJetstreamDispatcher, msgHandler, true, consOptions);
        nc.flush(Duration.ofMillis(50));
    } catch (final Exception e) {
        log.error("createConsumer exception " + e.getMessage());
        return null;
    }
    log.info("createConsumer done");

jnats version: 2.13.1
nats servers: 2.6.1

Could you please help me figure out what I'm missing in order to make work subscriptions, belonging to the same queue, sharing the messages and not just receive a broadcast for each of them?

BR,
Riccardo

[scottf]

@rpuppo converted this to a discussion

[scottf]

So your code looks ok. But is this the entire code? what happens after that last log.info call? If that's the end of your program, well the program will end. Even though there is another thread running, the main program still ends.
If that's not the problem get back to me and also have a look here: https://github.com/nats-io/nats.java/blob/main/src/examples/java/io/nats/examples/jetstream/NatsJsPushSubQueueDurable.java

[scottf]

@rpuppo Your consumer must be durable. Also, remove setting the deliver subject. Check out this example: https://github.com/nats-io/nats.java/blob/main/src/examples/java/io/nats/examples/jetstream/NatsJsPushSubQueueDurable.java

I would also consider creating the consumer ahead of time and binding to it. There is a bind example here: https://github.com/nats-io/nats.java/blob/main/src/examples/java/io/nats/examples/jetstream/NatsJsPushSubBindDurable.java

PushSubscribeOptions consOptions = PushSubscribeOptions.bind(config.streamName, config.durableName);

[rpuppo]

Thanks @scottf for your precious help!
Now it works fine , by doing exactly as you suggested.
Jetstream well balances across multiple subscriptions for the same consumer queue - deliver group.

Yet two small queries about:

• I saw you said to remove the deliver subject, but I get an error if I remove it (a consumer created ahead of time and then bind) --> "Consumer is already configured as a pull consumer"
• Is there a way to check, via nats utilities/nats box the number of subscriptions opened for an active consumer on a stream?
  I could not find something like that in "nats stream info", "nats stream report", "nats consumer info", "nats consumer report"

Thanks for your availability in advance.

BR,
Riccardo

[scottf]

So in your example, you created the consumer as part of the subscribe. In this case the deliver subject is unnecessary, the subscribe code will make one up. If you make the consumer ahead of time, you must set the deliver subject or else the consumer will be a pull consumer.

[rpuppo]

Clear @scottf , thanks for heads up!
About a utility to discover the number of instances consuming and participating to the same queue, do you have something?

[scottf]

I don't think so, I think I recently saw something about this in the slack channel. I would ask there or maybe in the https://github.com/nats-io/natscli