Pull consumers are not re-subscribing on Reconnect of NATS connection

[Scorpio1987]

Hi,

I am creating a consumer as shown in the example code

https://github.com/nats-io/nats.java/blob/main/src/examples/java/io/nats/examples/jetstream/simple/MessageConsumerExample.java

But i see that on re-connection of NATS server, the consumer is not getting resubscribed and i keep getting heartbeatAlarm errors. Is this the expected behavior and do i need to handle the reconnect event and re-create the consumer again?

Thanks,
Gopinadh.

[scottf]

What version of the client. There is a new bug introduced in 2.21.3

[Scorpio1987]

I am using client version 2.20.5

[Scorpio1987]

Here is sample consumer code i am using

import io.nats.client.*
import io.nats.client.api.*

fun kvStoreConsumer(streamName: String) {
    val options = Options.builder().server("nats://localhost:4222").build()
    val nc = Nats.connect(options)
    val js = nc.jetStream()
    val consumeOptions = ConsumeOptions.builder()
        .batchSize(15000)
        .expiresIn(1000)
        .build()

    val streamContext = nc.getStreamContext(streamName)
    val consumerContext = streamContext.createOrUpdateConsumer(
        ConsumerConfiguration.builder()
            .name("my-test-consumer")
            .deliverPolicy(DeliverPolicy.LastPerSubject)
            .ackPolicy(AckPolicy.None)
            .filterSubjects(">")
            .build())

    val consumer = consumerContext.consume(consumeOptions) { msg ->
        println("Received message: $msg")
    }

    while (!consumer.isFinished) {
        sleep(1000)
    }
}

[scottf]

It works fine for me. Try to upgrade to the latest library.

1. start the program
2. let it run for a few seconds, watch the output
3. kill the server
4. restart the server
5. watch the output

import io.nats.client.*;
import io.nats.client.api.*;

public class Test1374 {
    public static void main(String[] args) throws Exception {
        Options options = new Options.Builder()
            .server(Options.DEFAULT_URL)
            .build();

        String stream = "stream1347";
        String subject = "subject1347";
        String conName = "con1347";
        try (Connection nc = Nats.connect(options)) {
            // delete the stream for a fresh start.
            // catch exception we don't care if it didn't exist
            try {
                nc.jetStreamManagement().deleteStream(stream);
            }
            catch (Exception ignore) {}

            // make the stream
            nc.jetStreamManagement()
                .addStream(
                    StreamConfiguration.builder()
                        .name(stream)
                        .subjects(subject)
                        .storageType(StorageType.File) // HAS TO BE FILE
                        .build());

            // consume options are not necessary for the example
            ConsumeOptions consumeOptions = ConsumeOptions.builder()
                .batchSize(15000)
                .expiresIn(1000)
                .build();

            StreamContext streamContext = nc.getStreamContext(stream);
            ConsumerContext consumerContext = streamContext.createOrUpdateConsumer(
                ConsumerConfiguration.builder()
                    .name(conName)
                    .deliverPolicy(DeliverPolicy.LastPerSubject)
                    .ackPolicy(AckPolicy.None)
                    .filterSubjects(">")
                    .build());

            MessageHandler handler = msg -> System.out.println("Received message: " + new String(msg.getData()));

            //noinspection resource this should really be done in a try-resource, but it matches the original code.
            MessageConsumer mcon = consumerContext.consume(consumeOptions, handler);

            JetStream js = nc.jetStream();
            int x = 0;

            //noinspection InfiniteLoopStatement just run until the user kills the program.
            while (true) {
                try {
                    String data = "Data" + (++x);
                    js.publish(subject, data.getBytes());
                    System.out.println("Published message: " + data);
                    //noinspection BusyWait
                    Thread.sleep(250);
                }
                catch (Exception e) {
                    // publish can fail during disconnect
                    // try again in a bit
                    //noinspection BusyWait
                    Thread.sleep(1000);
                }
            }
        }
    }
}

[scottf]

As far as I can tell the current release works and I've made sure in the upcoming release.