ramonberrutti
commented
1 day ago Hey, I wrote a small test with the issue:
Looking into the code, it seems that when the message is added to the Redelivery List, it is still in the pending list. (Also the Timeout is updated)
func TestConsumerNumAckPending(t *testing.T) {
s := RunBasicJetStreamServer(t)
defer s.Shutdown()
nc, js := jsClientConnect(t, s)
defer nc.Close()
_, err := js.AddStream(&nats.StreamConfig{
Name: "TEST",
Subjects: []string{"test"},
})
require_NoError(t, err)
_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
Durable: "dur",
AckPolicy: nats.AckExplicitPolicy,
AckWait: 2 * time.Second,
})
require_NoError(t, err)
for i := 0; i < 5; i++ {
_, err := js.Publish("test", nil)
require_NoError(t, err)
}
info, err := js.ConsumerInfo("TEST", "dur")
require_NoError(t, err)
require_Equal(t, info.NumAckPending, 0)
require_Equal(t, info.NumPending, 5)
req := JSApiConsumerGetNextRequest{Batch: 1}
reqb, _ := json.Marshal(req)
sub := natsSubSync(t, nc, nats.NewInbox())
err = nc.PublishRequest("$JS.API.CONSUMER.MSG.NEXT.TEST.dur", sub.Subject, reqb)
require_NoError(t, err)
_ = natsNexMsg(t, sub, time.Second)
info, err = js.ConsumerInfo("TEST", "dur")
require_NoError(t, err)
require_Equal(t, info.NumAckPending, 1)
require_Equal(t, info.NumPending, 4)
checkFor(t, 10*time.Second, 1*time.Second, func() error {
info, err := js.ConsumerInfo("TEST", "dur")
if err != nil {
return err
}
if info.NumAckPending != 0 {
return fmt.Errorf("Expected no ack pending, got %d", info.NumAckPending)
}
if info.NumPending != 5 {
return fmt.Errorf("Expected 5 pending, got %d", info.NumPending)
}
return nil
})
}
derekcollison
MauriceVanVeen
Observed behavior
If an application terminates before acking its messages from a durable Jetstream Consumer and no further batches of messages are requested from the Consumer, num_ack_pending in ConsumerInfo never becomes zero.
Expected behavior
num_ack_pending should become zero after the JetStream Consumer's AckWait period expires on all outstanding messages.
Server and client version
nats-server 2.10.18
Host environment
Any. Apple M3 Pro.
Steps to reproduce
1. Create a stream and durable consumer w/ explicit ack & ackwait.
2 Receive a message without acknowledging it.
3 Poll ConsumerInfo
Notice that last activity is greater than the ack wait period yet num_ack_pending still shows there's an outstanding message.
The message is immediately ready for redelivery if another request comes along to pull messages.
It seems the ConsumerInfo counters are only advanced by requests which mutate the underlying stream.
As a workaround, I tried to request a zero-sized batch.
natscli appears to no-op on the client side for such a silly request.
I crafted the batch request manually, but nats-server clamps my batchsize back up to 1 and this counts against the delivery count of the sacrificial message.