Reconnect after a jetstream server restart

[amitza]

I'm having trouble understanding how a reconnection of the publisher and consumer works in case of a server restart.
I have one publisher, one consumer, and one stream. Using the simplified API:

Stream creation:

StreamConfiguration sc = StreamConfiguration.Builder()
				.WithName(stream)
				.WithStorageType(StorageType.File)
				.WithRetentionPolicy(RetentionPolicy.Limits)
				.WithMaxMessages(MaxMessages)
				.WithSubjects(subjects)
				.Build();

			StreamInfo streamInfo = jetStreamManagement.AddStream(sc);

Producing a message:

return _streamContext.PublishAsync(message: new Msg(subject: key, data: Serializer.Serialize(message)));

Consumer creation and consuming messages:

var context = _connection.CreateStreamContext(Stream);
				var consumerConfig = ConsumerConfiguration.Builder()
				.WithDeliverSubject(Subject)
				.WithName(ConsumerName)
				.WithDurable(ConsumerName)
				.Build();
				var consumerContext = context.CreateOrUpdateConsumer(consumerConfig);
				_logger.LogInformation($"A consumer was created on stream {Stream}. The consumer info is {consumerContext.GetConsumerInfo()}");

				_logger.LogInformation($"{consumerContext.ConsumerName} starts consuming. Interested in subject: {Subject} and participating in delivery group: {ConsumerName}");
				var consumerOptions = ConsumeOptions.Builder()
				.WithBatchSize(BatchSize)
				.Build();
				var messageConsumer = consumerContext.Consume((s, e) =>
				{
					_logger.LogTrace($"{consumerContext.ConsumerName} recived a new message {e.Message}");
					var decoded = _serializer.Deserialize(e.Message.Data);
					onMessage(new NATSMessageEnvelope<T>
					{
						RawMsg = e.Message,
						Data = decoded
					});
					if (ackMessages)
					{
						e.Message.Ack();
					}
				}, consumerOptions);

After a server restart I get the following error from the consumer:

PullStatusError, Connection: 17, Subscription: 2, ConsumerName:Consumer, Status: Status 409 Server Shutdown
PullStatusError, Connection: 17, Subscription: 2, ConsumerName:Consumer, Status: Status 409 Server Shutdown
PullStatusError, Connection: 17, Subscription: 2, ConsumerName:Consumer, Status: Status 409 Server Shutdown
PullStatusError, Connection: 17, Subscription: 2, ConsumerName:Consumer, Status: Status 409 Server Shutdown
DisconnectedEvent, Connection: 17
ReconnectedEvent, Connection: 11

And from the publisher:

DisconnectedEvent, Connection: 12
Unhandled exception. ReconnectedEvent, Connection: 13
System.NotSupportedException: Stream does not support seeking.
   at System.IO.BufferedStream.get_Position()
   at NATS.Client.Connection.PublishImpl(String subject, String reply, MsgHeader inHeaders, Byte[] data, Int32 offset, Nullable`1 inCount, Boolean flushBuffer)
   at NATS.Client.Connection.RequestAsyncImpl(String subject, MsgHeader headers, Byte[] data, Int32 offset, Nullable`1 count, Int32 timeout, CancellationToken token)
   at NATS.Client.JetStream.JetStream.PublishAsyncInternal(String subject, Byte[] data, MsgHeader hdr, PublishOptions options)
   at Program.<Main>$(String[] args) in C:\Users\aamit\source\repos\SampleJsonPublisher\SampleJsonPublisher\Program.cs:line 68
   at Program.<Main>(String[] args)

And messages just stop arriving.

Consumer info:

Information for Consumer policies > Consumer created 2023-07-13T11:43:33+03:00

Configuration:

                Name: Consumer
           Pull Mode: true
      Deliver Policy: All
          Ack Policy: Explicit
            Ack Wait: 30s
       Replay Policy: Instant
     Max Ack Pending: 1,000
   Max Waiting Pulls: 512

State:

   Last Delivered Message: Consumer sequence: 656 Stream sequence: 656
     Acknowledgment floor: Consumer sequence: 656 Stream sequence: 656
         Outstanding Acks: 0 out of maximum 1,000
     Redelivered Messages: 0
     Unprocessed Messages: 12
            Waiting Pulls: 0 of maximum 512

Stream info:

Information for Stream policies created 2023-07-13 11:43:18

             Subjects: policies
             Replicas: 1
              Storage: File

Options:

            Retention: Limits
     Acknowledgements: true
       Discard Policy: Old
     Duplicate Window: 2m0s
    Allows Msg Delete: true
         Allows Purge: true
       Allows Rollups: false

Limits:

     Maximum Messages: 10,000
  Maximum Per Subject: unlimited
        Maximum Bytes: unlimited
          Maximum Age: unlimited
 Maximum Message Size: unlimited
    Maximum Consumers: unlimited


State:

             Messages: 668
                Bytes: 34 KiB
             FirstSeq: 1 @ 2023-07-13T08:43:18 UTC
              LastSeq: 668 @ 2023-07-13T09:39:22 UTC
     Active Consumers: 1
   Number of Subjects: 1

I'm probobly missing something I'm just not sure what.

Additional details:

dotnet version: 6
Nets.Client version: 1.0.6

[scottf]

I'll look into this. What version of the server? Also, please update to 1.0.7

[amitza]

2023-07-13 12:39:27 [1] 2023/07/13 09:39:27.182016 [INF]   Version:  2.9.19
2023-07-13 12:39:27 [1] 2023/07/13 09:39:27.182019 [INF]   Git:      [f4f3cce]
2023-07-13 12:39:27 [1] 2023/07/13 09:39:27.182020 [INF]   Name:     NCHGWVF6GTHDWNDTEZUQZY4DRXDJSNDOZACLEKLE2NEUE674JG7LYN3G
2023-07-13 12:39:27 [1] 2023/07/13 09:39:27.182022 [INF]   Node:     piMpmo2q
2023-07-13 12:39:27 [1] 2023/07/13 09:39:27.182023 [INF]   ID:       NCHGWVF6GTHDWNDTEZUQZY4DRXDJSNDOZACLEKLE2NEUE674JG7LYN3G

thanks!

[scottf]

How many servers in your cluster. Publishing should reconnect right away to another server, but there is really no recovery for either publish or consume if there are no more servers that it can connect to. You will just have to have your application handle the exceptions and use the error event handlers to be aware of what's going on.

[scottf]

You can also use lame duck mode on the server to help the clients exit more gracefully, but without a cluster, there is not recovery.

[amitza]

I'm using one nats server only. I'm expecting it to reconnect after I restart the nats server. Is this something I need to take care of on my own? The server is not available for 1 second and then comes up immediately.