fix(nats-jetstream): correctly count messages that should be redelivered (waiting for ack) towards keda value (#3809)

toniopelo · web-flow · commit 971ab9485888 · 2022-11-09T07:51:30.000Z
* fix: keda now include the messages that should be retried in the count of pending messages used for scaling

Signed-off-by: Antoine Laffargue &lt;antoine.laffargue@gmail.com&gt;

* chore: update changelog

Signed-off-by: Antoine Laffargue &lt;antoine.laffargue@gmail.com&gt;

Signed-off-by: Antoine Laffargue &lt;antoine.laffargue@gmail.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -67,6 +67,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **General:** Respect optional parameter inside envs for ScaledJobs ([#3568](https://github.com/kedacore/keda/issues/3568))
 - **Azure Blob Scaler** Store forgotten logger ([#3811](https://github.com/kedacore/keda/issues/3811))
 - **Prometheus Scaler:** Treat Inf the same as Null result ([#3644](https://github.com/kedacore/keda/issues/3644))
+- **NATS Jetstream:** Correctly count messages that should be redelivered (waiting for ack) towards keda value ([#3787](https://github.com/kedacore/keda/issues/3787))
 
 ### Deprecations
 
diff --git a/pkg/scalers/nats_jetstream_scaler.go b/pkg/scalers/nats_jetstream_scaler.go
@@ -195,7 +195,7 @@ func (s *natsJetStreamScaler) getMaxMsgLag() int64 {
 
 	for _, consumer := range s.stream.Consumers {
 		if consumer.Name == consumerName {
-			return int64(consumer.NumPending)
+			return int64(consumer.NumPending + consumer.NumAckPending)
 		}
 	}
 	return s.stream.State.LastSequence

Original file line number	Diff line number	Diff line change
`@@ -195,7 +195,7 @@ func (s *natsJetStreamScaler) getMaxMsgLag() int64 {`
`195`	`195`
`196`	`196`	`for _, consumer := range s.stream.Consumers {`
`197`	`197`	`if consumer.Name == consumerName {`
`198`		`- return int64(consumer.NumPending)`
	`198`	`+ return int64(consumer.NumPending + consumer.NumAckPending)`
`199`	`199`	`}`
`200`	`200`	`}`
`201`	`201`	`return s.stream.State.LastSequence`