How to resume reading from last read message ? My consumer always reads from beginning

[kharabela]

output always starts from beginning

2024/07/23 09:27:34 The connection is Connected
2024/07/23 09:27:34 Messages - Pending: 1, Waiting: 0, Ack Pending: 19, Consumer Name: AccountSubjects, Stream: Subjects
2024/07/23 09:27:34 Received Message: Subjects.AccountInfo

Here is my sample consumer code.

`package main

import (
"context"
"encoding/json"
"log"
"os"
"os/signal"
"time"

"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"

)

func main() {

nc, err := nats.Connect(nats.DefaultURL, nats.ReconnectWait(10*time.Second))

if err != nil {
	log.Fatal(err)
}
defer nc.Drain()
/*Drain is a safe way to ensure all buffered messages that were published
are sent and all buffered messages received on a subscription are processed
being closing the connection.*/

ctx := context.Background()

getStatusTxt := func(nc *nats.Conn) string {
	switch nc.Status() {
	case nats.CONNECTED:
		return "Connected"
	case nats.CLOSED:
		return "Closed"
	default:
		return "Other"
	}
}
log.Printf("The connection is %v\n", getStatusTxt(nc))

// Use the JetStream context to produce and consumer messages that have been persisted.

//js, err := nc.JetStream(nats.PublishAsyncMaxPending(10000)) //ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
js, err := jetstream.New(nc)
if err != nil {
	log.Fatal(err)
}

// Create a stream if it does not exist
stream, err := js.Stream(ctx, "Subjects")
if err != nil {
	log.Fatal(err)
}

consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
	Name:      "AccountSubjects",
	Durable:   "AccountSubjects", //FilterSubject:Subjects.Account,AckPolicy:jetstream.AckExplicitPolicy
	AckPolicy: jetstream.AckExplicitPolicy,
})

if err != nil {
	log.Fatal(err)
}

ci, err := consumer.Info(ctx)
if err != nil {
	log.Fatal(err)
}

log.Printf("Messages - Pending: %d, Waiting: %d, Ack Pending: %d, Consumer Name: %s, Stream: %s", ci.NumPending, ci.NumWaiting, ci.NumAckPending, ci.Name, ci.Stream)

account := AccountInfoRecord{}
cctx, err := consumer.Consume(func(msg jetstream.Msg) {
	log.Printf("Received Message: %s", string(msg.Subject()))
	log.Printf("Received Message Data: %s", string(msg.Data()))

	msg.Ack()
	err = json.Unmarshal([]byte(msg.Data()), &account)
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("Consumer Data =>  Page: %s  -  Name: %s\n", account.PageName, account.Element.FirstName)
})

if err != nil {
	log.Fatal(err)
}

defer cctx.Stop()

//stream.DeleteConsumer(ctx, "AccountSubjects")

//gracefully shutdown
quit := make(chan os.Signal, 1)
signal.Notify(quit, os.Interrupt)
<-quit

}
`

[amitamit281]

Hi,
it is not reading from the beginning but messages are being redelivered by the server because those are not being acked.

i test your code, it works fine for me. I commented out consumer data logger & Unmarshal line because i don't have AccountInfoRecord class. That means somehome msg.Ack() line being skipped. Hence not being acked.

[kharabela]

Thanks a follow up question.How to identify last acknowledged message?

[amitamit281]

@kharabela you can check the stream sequence number under Acknowledgment Floor.

[kharabela]

@kharabela you can check the stream sequence number under Acknowledgment Floor.

I am using this sample code
Here is my code to read last acknowledge offset
log.Printf(" Last Delivered Sequence: %v, Last Ack Offset Consumer: %d,Last Ack Offset Stream: %d, Stream: %s", ci.Delivered, ci.AckFloor.Consumer, ci.AckFloor.Stream, ci.Stream)

Output:
Last Delivered Sequence: {169 71 },Last Ack Offset Consumer: 169,Last Ack Offset Stream: 71

calling this method

func printStreamState(ctx context.Context, stream jetstream.Stream) {
	info, _ := stream.Info(ctx)
	b, _ := json.MarshalIndent(info.State, "", " ")
	log.Println("inspecting stream info")
	log.Println(string(b))
}

Output:
2024/08/22 11:24:57 inspecting stream info
2024/08/22 11:24:57 {
"messages": 71,
"bytes": 52767,
"first_seq": 1,
"first_ts": "2024-06-17T05:17:14.1961524Z",
"last_seq": 71,
"last_ts": "2024-08-22T05:21:35.1089305Z",
"consumer_count": 2,
"deleted": null,
"num_deleted": 0,
"num_subjects": 1,
"subjects": null
}

My intent here is to read new messages for last_seq:71 onwards
Here is my code to do that

consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		Name:          "AccountSubjects",
		Durable:       "AccountSubjects", //FilterSubject:Subjects.Account,AckPolicy:jetstream.AckExplicitPolicy
		AckPolicy:     jetstream.AckExplicitPolicy,
		DeliverPolicy: jetstream.DeliverByStartSequencePolicy, 
		OptStartSeq:   71,
		MaxAckPending: 1000,
	})

cctx, err := consumer.Consume(func(msg jetstream.Msg) {
		log.Printf("Received Message: %s", string(msg.Subject()))
		log.Printf("Received Message Data: %s", string(msg.Data()))

		msg.Ack()
		err = json.Unmarshal([]byte(msg.Data()), &account)
		if err != nil {
			log.Fatal(err)

		}
		log.Printf("Consumer Data =>  Page: %s  -  Name: %s\n", account.PageName, account.Element.FirstName)
	})

However I am getting this error

2024/08/22 11:38:42 nats: API error: code=500 err_code=10012 description=deliver policy can not be updated
exit status 1

Also, Is there a way to dynamically update OptStartSeq value rather than hard coding?

[amitamit281]

@kharabela It doesn't support updating the delivery policy, so you'll need to recreate the consumer. However, if you can identify why the messages aren't being acknowledged, you might not need to recreate it. NATS will only deliver messages from that point onward, not from the beginning.