fix: add max pending ACK to the nats consumer configuration (#219)

kre-go/config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"os"
+	"strconv"
 	"strings"
 
 	"github.com/konstellation-io/kre/libs/simplelogger"
@@ -35,13 +36,19 @@ type ConfigNATS struct {
 	KeyValueStoreWorkflowName string
 	KeyValueStoreNodeName     string
 	MongoWriterSubject        string
+	MaxPendingAck             int
 }
 
 type InfluxDB struct {
 	URI string
 }
 
 func NewConfig(logger *simplelogger.SimpleLogger) Config {
+	maxPendingAck, err := strconv.Atoi(getOptCfgFromEnv(logger, "KRT_MAX_PENDING_ACK"))
+	if err != nil {
+		maxPendingAck = -1
+	}
+
 	return Config{
 		WorkflowName: getCfgFromEnv(logger, "KRT_WORKFLOW_NAME"),
 		RuntimeID:    getCfgFromEnv(logger, "KRT_RUNTIME_ID"),
@@ -59,6 +66,7 @@ func NewConfig(logger *simplelogger.SimpleLogger) Config {
 			KeyValueStoreWorkflowName: getCfgFromEnv(logger, "KRT_NATS_KEY_VALUE_STORE_WORKFLOW"),
 			KeyValueStoreNodeName:     getCfgFromEnv(logger, "KRT_NATS_KEY_VALUE_STORE_NODE"),
 			MongoWriterSubject:        getCfgFromEnv(logger, "KRT_NATS_MONGO_WRITER"),
+			MaxPendingAck:             maxPendingAck,
 		},
 		MongoDB: MongoDB{
 			Address:     getCfgFromEnv(logger, "KRT_MONGO_URI"),
@@ -86,6 +94,7 @@ func getOptCfgFromEnv(logger *simplelogger.SimpleLogger, name string) string {
 		logger.Infof("The %q env var is missing", name)
 		return ""
 	}
+
 	return val
 }
 

kre-go/main.go

@@ -101,6 +101,7 @@ func Start(handlerInit HandlerInit, defaultHandler Handler, handlersOpt ...map[s
 			nats.Durable(consumerName),
 			nats.ManualAck(),
 			nats.AckWait(22*time.Hour),
+			nats.MaxAckPending(cfg.NATS.MaxPendingAck),
 		)
 		if err != nil {
 			logger.Errorf("Error subscribing to NATS subject %s: %s", subject, err)