Re-subscriptions after reconnect are not honored

[ewirch]

Defect

We use the Java client library to create a pull-subscription and listen on it. When NATS server is restarted, the lib reconnects and even re-sends the subcriptions, but NATS won't send any messages to the client any more.

• Included nats-server -DV output
• Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)

nats-server -DV

I'm running docker.io/library/nats@sha256:58e483681983f87fdc738e980d582d4cc7218914c4f0056d36bab0c5acfc5a8b locally. Executing the above command in the running container gives me:

Starting nats-server
  Version:  2.9.15
  Git:      [b91fa85]
  Go build: go1.19.6
  Name:     NBDPZPHME36GTJH53Q3RBRJCQAKTAWASAKBCYIOAPRV5PYOBEQHP6UA7
  ID:       NBDPZPHME36GTJH53Q3RBRJCQAKTAWASAKBCYIOAPRV5PYOBEQHP6UA7
Created system account: "$SYS"
Error listening on port: 0.0.0.0:4222, "listen tcp 0.0.0.0:4222: bind: address already in use"

Versions of nats-server and affected client libraries used:

nats-server: 2.9.15
client: io.nats:jnats:2.16.9

OS/Container environment:

Podman 4.4.1 on Arch Linux.

Steps or code to reproduce the issue:

Starting with this basic listener:

package test.project;

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

import java.io.IOException;
import java.time.Duration;
import java.util.Optional;

public class NatsSubscriber {

    private static final String STREAM_NAME = "stream-1";

    public static void main(String[] args) throws InterruptedException {
        new NatsSubscriber().run();
    }

    private void run() throws InterruptedException {
        try (var connection = createConnection()) {
            createStreamIfNotExists(connection);
            var subscription = createSubscription(connection);
            subscription.pull(10);
            while (true) {
                try {
                    System.out.println("listening...");
                    var message = subscription.nextMessage(Duration.ZERO);
                    if (message != null) {
                        System.out.println("received msg: "+ new String(message.getData()));
                        message.ack();
                    }
                } catch (JetStreamStatusException e) {
                    // ignore "Unknown or unprocessed status message: Server Shutdown"
                }
            }
        }
    }

    public Connection createConnection() {
        var connectionOptions = new Options.Builder()
                .server("nats://localhost:4222")
                .connectionName("nats-subscriber-1")
                .errorListener(new ErrorListenerLoggerImpl())
                .noNoResponders()
                .build();

        try {
            return Nats.connect(connectionOptions);
        } catch (IOException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private JetStreamSubscription createSubscription(Connection connection) {
        var subscriptionOptions = PullSubscribeOptions.builder()
                .configuration(getConsumerOptions())
                .stream(STREAM_NAME)
                .build();

        try {
            return connection.jetStream().subscribe(STREAM_NAME, subscriptionOptions);
        } catch (IOException | JetStreamApiException e) {
            throw new RuntimeException(e);
        }
    }

    ConsumerConfiguration getConsumerOptions() {
        return ConsumerConfiguration.builder()
                .deliverPolicy(DeliverPolicy.All)
                .ackPolicy(AckPolicy.Explicit)
                .durable("consumer-1")
                .build();
    }

    private void createStreamIfNotExists(Connection connection) {
        try {
            var streamInfo = getStreamInfo(connection);

            if (streamInfo.isEmpty()) {
                createNewStream(connection);
            }
        } catch (IOException | JetStreamApiException e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }


    private Optional<StreamInfo> getStreamInfo(Connection connection) throws IOException, JetStreamApiException {
        try {
            return Optional.of(connection.jetStreamManagement().getStreamInfo(STREAM_NAME));
        } catch (JetStreamApiException ex) {
            if (ex.getErrorCode() == 404) {
                return Optional.empty();
            }
            throw ex;
        }
    }

    private void createNewStream(Connection connection) throws JetStreamApiException, IOException {
        var streamConfiguration = getStreamConfiguration();
        connection.jetStreamManagement().addStream(streamConfiguration);
    }

    private StreamConfiguration getStreamConfiguration() {
        return StreamConfiguration.builder()
                .name(STREAM_NAME)
                .storageType(StorageType.File)
                .retentionPolicy(RetentionPolicy.WorkQueue)
                .replicas(1)
                .build();
    }
}

(full project: test-project.zip )

Start nats-server:

podman image pull docker.io/library/nats@sha256:58e483681983f87fdc738e980d582d4cc7218914c4f0056d36bab0c5acfc5a8b
podman run -p 4222:4222 -p 8222:8222 -p 6222:6222 --name nats-server -d docker.io/nats --jetstream --trace

(you can also simply replace podman by docker)
Follow the logs in the background:

podman logs -f <container-id>

Start the sample project above. App waits for messages after subscribing to stream:

listening...

Observe in nats-server output, that nats received the subscibe:

[TRC] 10.0.2.100:60886 - cid:60 - <<- [SUB _INBOX.KQyEsJiBiM22h3CgB48uTJ 2]

Check out the consumer info:

$ nats consumer info stream-1 consumer-1
State:
    ...
    Waiting Pulls: 1 of maximum 512

(I removed irrelevant lines)
So there is a pull registered.

Publish a message:

nats publish stream-1 test

Observe in app output, that message was received:

received msg: test

Now restart nats-server:

podman container restart <container-id>

Again, observe in nats-server output that nats receives subscriptions after re-connect:

[TRC] 10.0.2.100:57696 - cid:41 - <<- [SUB _INBOX.KQyEsJiBiM22h3CgB48uTJ 2]
[TRC] 10.0.2.100:57696 - cid:41 - <<- [SUB _INBOX.KQyEsJiBiM22h3CgB48uAb.* 1]

Check out the consumer info:

$ nats consumer info stream-1 consumer-1
State:
    ...
    Waiting Pulls: 0 of maximum 512

Nats has no waiting pull any more!

Publish a message:

nats publish stream-1 test

Observe, that app does not receive the message (nats does not deliver it). When you restart the app, message will be received.

Expected result:

Messages are received after re-connect.

Actual result:

Messages are not received after re-connect.

[ewirch]

This is a serious problem for our production environment with its 37 streams and their 610 consumers. Any workarounds we could implement, until the bug is fixed?

[derekcollison]

@scottf could you respond here. Thanks.

[kchrusciel]

@scottf @derekcollison friendly reminder.

[scottf]

@ewirch Sorry for the delay getting on this. I have some questions.
What does the docker / server not restarting have to do with the problem, or is it the root cause? Probably nothing, just checking.
I know the java client reconnects for pulls, even with 2.16.9.
I also know that the if the consumer leader is on the server that is down, pull gets nothing, heartbeats will help detect this.
Can you upgrade to 2.16.10? 2.16.10 is better than 2.16.9 but pull heartbeats are in both.
Can you change your pull to something with an expire and and idle heartbeat? So instead of this

subscription.pull(10);

try something like this:

PullRequestOptions pro = PullRequestOptions.builder(10)
    .expiresIn(10_000)
    .idleHeartbeat(1_000)
    .build();
subscription.pull(pro);

Another thing to look into is this set of example code that runs against 2.16.10. It demonstrates how to use heartbeats, the error listener and what to expect in a few different situations, including when a server comes down, especially the ConnectionLostWhilePullActive.java class
https://github.com/nats-io/java-nats-examples/tree/main/error-and-heartbeat-experiments/src/main/java/io/nats

[ewirch]

Hi @scottf. Thanks for the hints. Sadly they did not lead to a workaround, which we could use.

What does the docker / server not restarting have to do with the problem

There is no "not restarting". The NATS server restarts gracefully. I never claimed the server would fail to restart. I'm restarting the server to test the reconnection behavior of the library.

Here are my findings following your tips:

Trying

• docker.io/library/nats:2.9.16-alpine
• io.nats:jnats:2.16.10

without code modifications shows same behavior.

Using pull with heart beats and expire timeout shows interesting new behavior. subscription.nextMessage(Duration.ZERO) returns null when the pull/subscription (what's the correct term?) expires. I suppose this is expected. The subscriber calls nextMessage() in a loop. This second nextMessage() will block forever. No heart beats, no expire timeout, no waiting pull. Subscriber output:

listening...
listening...

NATS server output:

[TRC] 10.0.2.100:39456 - cid:10 - <<- [SUB _INBOX.12sb1qsvabW7kpfHYosjw8 2]
[TRC] 10.0.2.100:39456 - cid:10 - <<- [PUB $JS.API.CONSUMER.MSG.NEXT.stream-1.consumer-1 _INBOX.12sb1qsvabW7kpfHYosjw8 76]
[TRC] 10.0.2.100:39456 - cid:10 - <<- MSG_PAYLOAD: ["{\"batch\":10,\"max_bytes\":0,\"expires\":10000000000,\"idle_heartbeat\":1000000000}"]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [PING]
[TRC] 10.0.2.100:39456 - cid:10 - <<- [PONG]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 75 75]
[TRC] 10.0.2.100:39456 - cid:10 - ->> [HMSG _INBOX.12sb1qsvabW7kpfHYosjw8 2 82 82]

We see the first 10 heart beats, and then nothing. nats consumer info stream-1 consumer-1 also shows 0 waiting pulls.

Probably as expected as well? I initiated a pull, the pull expired. Do I need to pull again? Buf if this is the case, I'd expect nextMessage() to fail, so it would not listen forever as it does now.

I see no way to detect when the pull expired, so I simply move the pull() in to the loop:

var pro = PullRequestOptions.builder(10)
        .expiresIn(10_000)
        .idleHeartbeat(1_000)
        .build();
while (true) {
    try {
        System.out.println("listening...");
        subscription.pull(pro);
        var message = subscription.nextMessage(Duration.ZERO);
        if (message != null) {
            System.out.println("received msg: " + new String(message.getData()));
            message.ack();
        }
    } catch (JetStreamStatusException e) {
        ...
    }
}

This time, the pull is re-initiated, when it expires. We see a list of listening... outputs. Also nats consumer info stream-1 consumer-1 shows a waiting pull all the time. But this change seems wrong. In case a message is received, the pull is re-initiated as well. The code have no way to know if the old pull is still active or did expire. nats consumer info stream-1 consumer-1 shows multiple waiting pulls in that case.

It gets even more interesting when we restart the NATS server. The library gets in an infinite loop, printing the same line over and over again:

Apr 18, 2023 9:53:07 AM io.nats.client.impl.ErrorListenerLoggerImpl heartbeatAlarm
SEVERE: heartbeatAlarm, Connection: 9, Subscription: 1185142384, Consumer Name: consumer-1, lastStreamSequence: 0, lastConsumerSequence: 0

According to nats consumer info waiting pulls is 0. I think the multiple pull() invocations led to this problem. If there is only one pull() the lib does not get in to this state.

Another thing to look into is this set of example code that runs against 2.16.10. It demonstrates how to use heartbeats, the error listener and what to expect in a few different situations, including when a server comes down, especially the ConnectionLostWhilePullActive.java class

ConnectionLostWhilePullActive has the same problem. First, I had to change the stream storageType to File, because otherwise the stream (and consumer) are lost on restart. When running ConnectionLostWhilePullActive, there is one waiting pull per consumer. After NATS restart, there are 0 waiting pulls.

[scottf]

Ok I saw this and was just asking for clarification about whether this was to show a problem or not.

I'm running docker.io/library/nats@sha256:58e483681983f87fdc738e980d582d4cc7218914c4f0056d36bab0c5acfc5a8b locally. Executing the above command in the running container gives me:

Error listening on port: 0.0.0.0:4222, "listen tcp 0.0.0.0:4222: bind: address already in use"

[scottf]

@ewirch So next thing is this:

subscription.nextMessage(Duration.ZERO);

Zero (0) means block forever until there is a message. So even if the pull(pro) times out, the client will just sit there forever.

As far as the re-issuing the pulls, you are going to need to track the time somehow. I'll work on a way to surface information about raw-pulls when they complete, we just added them for simplification, but there is currently no way to get that info for the old api. The fetch is a good example of how to loop over a pull with expiration. Basically I'm tracking the time used and make sure it's more than the pull expiration or the full batch size, whichever comes first.

So for heartbeat alarms, when you get one via the error listener, this is the way you know the server is down or there is a likely non recoverable situation with the specific subscription. So there needs to be some logic in the code to recognize this state. At the moment I'm not sure if a pull will recover after a heartbeat warning, I'll have to put an example together.

Can I suggest you look into using the fetch or iterate api, or even looking into the new simplification api, there is a lot of work done for you?

[scottf]

So here are some examples using our new simplification api. (they currently experimental so signatures/ api can change)
https://github.com/nats-io/nats.java/tree/main/src/examples/java/io/nats/examples/jetstream/simple

So these will make the happy path much easier. The error listener still is required to know when there are connection / disconnection / reconnection events as well as heartbeat alarms. These indicators can be used to help recognize that an subscription is probably dead and needs to be discarded and re-made, possibly to another entire connection

[scottf]

Also this piece of code: https://github.com/nats-io/java-nats-examples/blob/main/error-and-heartbeat-experiments/src/main/java/io/nats/ConnectionLostWhilePullActive.java
is an attempt to have some code that can be executed in various ways (see comments in code) to see what happens with disconnections and reconnections.

[derekcollison]

I am going to move this to the java repo since its more about the java client, not the server.

[ewirch]

its more about the java client

Is it? Please review the original post. After a server restart, the java client correctly re-connects. Even sends subscription messages:

[TRC] 10.0.2.100:57696 - cid:41 - <<- [SUB _INBOX.KQyEsJiBiM22h3CgB48uTJ 2]
[TRC] 10.0.2.100:57696 - cid:41 - <<- [SUB _INBOX.KQyEsJiBiM22h3CgB48uAb.* 1]

but the server still considers the consumer to have 0 waiting pulls. So is this really as expected for the server?

[ewirch]

@scottf, do you think NATS server behavior described in original post (ignoring subscribes) is by design, or is it a bug?

[scottf]

@ewirch I'm think I understand. So if a server goes down and the client is disconnected, the pull is closed, which means even if the client reconnects, the pull is probably expired and you don't get messages.
I think by the nature of pull, the best thing to do is use the connection and event listeners and re-issue a pull (or even just close the one subscription and make another.) If you make another pull and there is one already in progress, they will both supply messages. If there isn't the original pull for whatever reason, i.e. it was lost in the disconnection or just expired, then you need a new pull anyway. Admittedly, the happy path with pull is great, and otherwise it can get tricky because in reality there isn't much the client can do. With a durable consumer, you at least have state and can restart a subscription where you left off.

[ewirch]

Thanks for the explanation, @scottf. In that case, I think the Java library pull API is flawed. The library detects a network problem and re-connects. It should check if the pull is still active, and either renew the pull or fail, so the lib caller can create a new pull.

[scottf]

For the low level api that allow you to do the pull directly, I don't agree. They are raw api after all. In regard to the macro api, like fetch/iterate/reader and simplification, there may be some level of recovery that we should have, but the question is how much? I can bring it up again about how much our macro api should recover.

[z0mb1ek]

@scottf Hi, i move to new api consumerContext.consume { message -> and get same issue. Is there any example how to handle it?

[scottf]

I'm working on this whole problem starting today and next week. Please be patient.

[rdkirony]

Any solution ?

[scottf]

Working on it right now.

[rdkirony]

I found a way to solve the problem. I don't know if it's the righ way, but i use the reconnect event to instanciate a new consumer and also i have to overwrite the heartBeatAlarm to close the garbage Dispatcher.

[scottf]

The heartbeat is the key. Could go as far as tracking connection events. There are a lot of variables to track, I'm debugging. Will have something by Monday.

[scottf]

@ewirch @kchrusciel @rdkirony Can you see if the latest snapshot (2.17.2-SNAPSHOT) solves your problems?

• gradle/maven instructions here in the readme https://github.com/nats-io/nats.java#using-gradle

There is a similar issue here to follow #997. I'm closing this one in favor of it. But here is the synopsis.

The work done in these PRs...

• Client side consumer recovery #1043
• Ordered Push don't cancel sub if trying to recover #1045

This affects the following consumers.

1. Durable Push Consumer
2. Ephemeral Push Consumer with a long enough inactive threshold to survive a server outage or disconnection
3. Ordered Push Consumer
4. Durable Pull Consumer when used with endless Simplification API
5. Ephemeral Pull Consumer with a long enough inactive threshold to survive a server outage or disconnection when used with endless Simplification API
6. Ordered Pull Consumer with endless Simplification

[dfilipovic-ib]

Hey, we had same problem, As i can see what happens is:

• in JetStreamReaderImpl ctor sub.pull(batchSize); is called initially to request initial batch from server
• there is a check for msg == null here that totally skips pulling new batch and this is what happens when you resubscribe after client reconncts on server shutdown - in endless msg==null loop:
  

  nats.java/src/main/java/io/nats/client/impl/NatsJetStreamPullSubscription.java

  Line 342 in 0e4f39c

  if (msg != null) {

• so either removing this check or calling sub.pull on resubscription only in case when the implementation is JetStreamReader here:
  

  nats.java/src/main/java/io/nats/client/impl/NatsConnection.java

  Line 438 in 0e4f39c

  this.subscribers.forEach((sid, sub) -> {

  
  fixes the issue.
• imo the first solution should be the fix, but not sure in which cases can the statement be true.

[scottf]

Please move to the newer simplification api. Examples can be found here: https://github.com/nats-io/nats.java/tree/main/src/examples/java/io/nats/examples/jetstream/simple

The reader api was not designed to be robust like simplification is.