Fix consumer limits (#4567)

If Stream has consumer limits set, creating consumer with defaults will
fail in most cases.

Test didn't catch this, as by default, old JS client sets ack policy to
`none`. If the policy is different, it will fail to create consumer with
defaults. We agreed that default Ack Policy should be `explicit`

 Signed-off-by: Tomasz Pietrek <tomasz@nats.io>

server/consumer.go

@@ -437,6 +437,12 @@ func setConsumerConfigDefaults(config *ConsumerConfig, streamCfg *StreamConfig,
 	if len(config.BackOff) > 0 {
 		config.AckWait = config.BackOff[0]
 	}
+	if config.MaxAckPending == 0 {
+		config.MaxAckPending = streamCfg.ConsumerLimits.MaxAckPending
+	}
+	if config.InactiveThreshold == 0 {
+		config.InactiveThreshold = streamCfg.ConsumerLimits.InactiveThreshold
+	}
 	// Set proper default for max ack pending if we are ack explicit and none has been set.
 	if (config.AckPolicy == AckExplicit || config.AckPolicy == AckAll) && config.MaxAckPending == 0 {
 		accPending := JsDefaultMaxAckPending
@@ -452,12 +458,6 @@ func setConsumerConfigDefaults(config *ConsumerConfig, streamCfg *StreamConfig,
 	if config.DeliverSubject == _EMPTY_ && config.MaxRequestBatch == 0 && lim.MaxRequestBatch > 0 {
 		config.MaxRequestBatch = lim.MaxRequestBatch
 	}
-	if config.MaxAckPending == 0 {
-		config.MaxAckPending = streamCfg.ConsumerLimits.MaxAckPending
-	}
-	if config.InactiveThreshold == 0 {
-		config.InactiveThreshold = streamCfg.ConsumerLimits.InactiveThreshold
-	}
 }
 
 // Check the consumer config. If we are recovering don't check filter subjects.

server/jetstream_test.go

@@ -21718,7 +21718,8 @@ func TestJetStreamConsumerDefaultsFromStream(t *testing.T) {
 
 	t.Run("InheritDefaultsFromStream", func(t *testing.T) {
 		ci, err := js.AddConsumer("test", &nats.ConsumerConfig{
-			Name: "InheritDefaultsFromStream",
+			Name:      "InheritDefaultsFromStream",
+			AckPolicy: nats.AckExplicitPolicy,
 		})
 		require_NoError(t, err)