Repository: nats-io/nats-server
Pull request #3130 (2022-05-17 21:23:04)

[ADDED] MQTT: Stream/Consumer replica count override
Ability to override the stream and consumers replica count, which is by default
determined based on the cluster size.

```
mqtt {
  port: 1883
  stream_replicas: 5
  consumer_replicas: 1
}
```

The above would allow *new* MQTT streams to be created with a replicas
factor of 5 (it will be an error if the cluster does not have that
many nodes, and error will occur at runtime when the first client
on a given account connects), and new consumers would be R=1.

The MQTT existing streams/consumers for an account are not modified.

The stream_replicas can also obviously be reduced to 1 for a cluster
of 3 nodes if one desire to have those streams as R=1.

A value of 0 or negative is considered letting the server pick
the value (from 1 to 3 depending on standalone/cluster size).

There is another property that allows the consumers to be created
with memory storage instead of file:
```
mqtt {
  ..
  consumer_memory_storage: true
}
```

Those new settings are global and apply to new streams/consumers
only.

Related to https://github.com/nats-io/nats-server/issues/3116

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/mqtt.go b/server/mqtt.go
index 25d96cb0..a371268d 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -147,6 +147,7 @@ const (
 	mqttJSAStreamLookup   = "SL"
 	mqttJSAStreamDel      = "SD"
 	mqttJSAConsumerCreate = "CC"
+	mqttJSAConsumerLookup = "CL"
 	mqttJSAConsumerDel    = "CD"
 	mqttJSAMsgStore       = "MS"
 	mqttJSAMsgLoad        = "ML"
@@ -226,7 +227,6 @@ type mqttAccountSessionManager struct {
 	sl         *Sublist                    // sublist allowing to find retained messages for given subscription
 	retmsgs    map[string]*mqttRetainedMsg // retained messages
 	jsa        mqttJSA
-	replicas   int
 	rrmLastSeq uint64        // Restore retained messages expected last sequence
 	rrmDoneCh  chan struct{} // To notify the caller that all retained messages have been loaded
 	sp         *ipQueue      // of uint64. Used for cluster-wide processing of session records being persisted
@@ -577,6 +577,10 @@ func validateMQTTOptions(o *Options) error {
 	if err := validatePinnedCerts(mo.TLSPinnedCerts); err != nil {
 		return fmt.Errorf("mqtt: %v", err)
 	}
+	if mo.ConsumerReplicas > 0 && mo.StreamReplicas > 0 && mo.ConsumerReplicas > mo.StreamReplicas {
+		return fmt.Errorf("mqtt: consumer_replicas (%v) cannot be higher than stream_replicas (%v)",
+			mo.ConsumerReplicas, mo.StreamReplicas)
+	}
 	return nil
 }
 
@@ -916,18 +920,21 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 
 	accName := acc.GetName()
 
+	opts := s.getOpts()
 	c := s.createInternalAccountClient()
 	c.acc = acc
 
 	id := string(getHash(s.Name()))
-	replicas := s.mqttDetermineReplicas()
+	replicas := opts.MQTT.StreamReplicas
+	if replicas <= 0 {
+		replicas = s.mqttDetermineReplicas()
+	}
 	qname := fmt.Sprintf("[ACC:%s] MQTT ", accName)
 	as := &mqttAccountSessionManager{
 		sessions:   make(map[string]*mqttSession),
 		sessByHash: make(map[string]*mqttSession),
 		sessLocked: make(map[string]struct{}),
 		flappers:   make(map[string]int64),
-		replicas:   replicas,
 		jsa: mqttJSA{
 			id:     id,
 			c:      c,
@@ -942,7 +949,6 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 
 	// The domain to communicate with may be required for JS calls.
 	// Search from specific (per account setting) to generic (mqtt setting)
-	opts := s.getOpts()
 	if opts.JsAccDefaultDomain != nil {
 		if d, ok := opts.JsAccDefaultDomain[accName]; ok {
 			if d != _EMPTY_ {
@@ -1037,49 +1043,88 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 		as.sessPersistProcessing(closeCh)
 	})
 
-	// Create the stream for the sessions.
-	cfg := &StreamConfig{
-		Name:       mqttSessStreamName,
-		Subjects:   []string{mqttSessStreamSubjectPrefix + as.domainTk + ">"},
-		Storage:    FileStorage,
-		Retention:  LimitsPolicy,
-		Replicas:   as.replicas,
-		MaxMsgsPer: 1,
-	}
-	if _, created, err := jsa.createStream(cfg); err == nil && created {
-		as.transferUniqueSessStreamsToMuxed(s)
-	} else if isErrorOtherThan(err, JSStreamNameExistErr) {
-		return nil, fmt.Errorf("create sessions stream for account %q: %v", acc.GetName(), err)
-	}
-
-	// Create the stream for the messages.
-	cfg = &StreamConfig{
-		Name:      mqttStreamName,
-		Subjects:  []string{mqttStreamSubjectPrefix + ">"},
-		Storage:   FileStorage,
-		Retention: InterestPolicy,
-		Replicas:  as.replicas,
-	}
-	if _, _, err := jsa.createStream(cfg); isErrorOtherThan(err, JSStreamNameExistErr) {
-		return nil, fmt.Errorf("create messages stream for account %q: %v", acc.GetName(), err)
-	}
-
-	// Create the stream for retained messages.
-	cfg = &StreamConfig{
-		Name:      mqttRetainedMsgsStreamName,
-		Subjects:  []string{mqttRetainedMsgsStreamSubject},
-		Storage:   FileStorage,
-		Retention: LimitsPolicy,
-		Replicas:  as.replicas,
-	}
-	si, _, err := jsa.createStream(cfg)
-	if isErrorOtherThan(err, JSStreamNameExistErr) {
-		return nil, fmt.Errorf("create retained messages stream for account %q: %v", acc.GetName(), err)
+	lookupStream := func(stream, txt string) (*StreamInfo, error) {
+		si, err := jsa.lookupStream(stream)
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				return nil, nil
+			}
+			return nil, fmt.Errorf("lookup %s stream for account %q: %v", txt, accName, err)
+		}
+		if opts.MQTT.StreamReplicas == 0 {
+			return si, nil
+		}
+		sr := 1
+		if si.Cluster != nil {
+			sr += len(si.Cluster.Replicas)
+		}
+		if replicas != sr {
+			s.Warnf("MQTT %s stream replicas mismatch: current is %v but configuration is %v for '%s > %s'",
+				txt, sr, replicas, accName, stream)
+		}
+		return si, nil
+	}
+
+	if si, err := lookupStream(mqttSessStreamName, "sessions"); err != nil {
+		return nil, err
+	} else if si == nil {
+		// Create the stream for the sessions.
+		cfg := &StreamConfig{
+			Name:       mqttSessStreamName,
+			Subjects:   []string{mqttSessStreamSubjectPrefix + as.domainTk + ">"},
+			Storage:    FileStorage,
+			Retention:  LimitsPolicy,
+			Replicas:   replicas,
+			MaxMsgsPer: 1,
+		}
+		if _, created, err := jsa.createStream(cfg); err == nil && created {
+			as.transferUniqueSessStreamsToMuxed(s)
+		} else if isErrorOtherThan(err, JSStreamNameExistErr) {
+			return nil, fmt.Errorf("create sessions stream for account %q: %v", accName, err)
+		}
+	}
+
+	if si, err := lookupStream(mqttStreamName, "messages"); err != nil {
+		return nil, err
+	} else if si == nil {
+		// Create the stream for the messages.
+		cfg := &StreamConfig{
+			Name:      mqttStreamName,
+			Subjects:  []string{mqttStreamSubjectPrefix + ">"},
+			Storage:   FileStorage,
+			Retention: InterestPolicy,
+			Replicas:  replicas,
+		}
+		if _, _, err := jsa.createStream(cfg); isErrorOtherThan(err, JSStreamNameExistErr) {
+			return nil, fmt.Errorf("create messages stream for account %q: %v", accName, err)
+		}
 	}
+
+	// This is the only case where we need "si" after lookup/create
+	si, err := lookupStream(mqttRetainedMsgsStreamName, "retained messages")
 	if err != nil {
-		si, err = jsa.lookupStream(mqttRetainedMsgsStreamName)
+		return nil, err
+	} else if si == nil {
+		// Create the stream for retained messages.
+		cfg := &StreamConfig{
+			Name:      mqttRetainedMsgsStreamName,
+			Subjects:  []string{mqttRetainedMsgsStreamSubject},
+			Storage:   FileStorage,
+			Retention: LimitsPolicy,
+			Replicas:  replicas,
+		}
+		// We will need "si" outside of this block.
+		si, _, err = jsa.createStream(cfg)
 		if err != nil {
-			return nil, fmt.Errorf("lookup retained messages stream for account %q: %v", acc.GetName(), err)
+			if isErrorOtherThan(err, JSStreamNameExistErr) {
+				return nil, fmt.Errorf("create retained messages stream for account %q: %v", accName, err)
+			}
+			// Suppose we had a race and the stream was actually created by another
+			// node, we really need "si" after that, so lookup the stream again here.
+			si, err = lookupStream(mqttRetainedMsgsStreamName, "retained messages")
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 
@@ -1097,15 +1142,14 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 
 	// Using ephemeral consumer is too risky because if this server were to be
 	// disconnected from the rest for few seconds, then the leader would remove
-	// the consumer, so even after a reconnect, we would not longer receive
+	// the consumer, so even after a reconnect, we would no longer receive
 	// retained messages. Delete any existing durable that we have for that
 	// and recreate here.
 	// The name for the durable is $MQTT_rmsgs_<server name hash> (which is jsa.id)
 	rmDurName := mqttRetainedMsgsStreamName + "_" + jsa.id
-	resp, err := jsa.deleteConsumer(mqttRetainedMsgsStreamName, rmDurName)
 	// If error other than "not found" then fail, otherwise proceed with creating
 	// the durable consumer.
-	if err != nil && (resp == nil || resp.Error.Code != 404) {
+	if _, err := jsa.deleteConsumer(mqttRetainedMsgsStreamName, rmDurName); isErrorOtherThan(err, JSConsumerNotFoundErr) {
 		return nil, err
 	}
 	ccfg := &CreateConsumerRequest{
@@ -1119,7 +1163,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 		},
 	}
 	if _, err := jsa.createConsumer(ccfg); err != nil {
-		return nil, fmt.Errorf("create retained messages consumer for account %q: %v", acc.GetName(), err)
+		return nil, fmt.Errorf("create retained messages consumer for account %q: %v", accName, err)
 	}
 
 	if lastSeq > 0 {
@@ -3557,6 +3601,10 @@ func (sess *mqttSession) processJSConsumer(c *client, subject, sid string,
 			FilterSubject:  mqttStreamSubjectPrefix + subject,
 			AckWait:        ackWait,
 			MaxAckPending:  maxAckPending,
+			MemoryStorage:  opts.MQTT.ConsumerMemoryStorage,
+		}
+		if r := opts.MQTT.ConsumerReplicas; r > 0 {
+			cc.Replicas = r
 		}
 		if err := sess.createConsumer(cc); err != nil {
 			c.Errorf("Unable to add JetStream consumer for subscription on %q: err=%v", subject, err)
diff --git a/server/opts.go b/server/opts.go
index 39ccbeea..3a32d5ea 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -411,6 +411,26 @@ type MQTTOpts struct {
 	// JetStream domain mqtt is supposed to pick up
 	JsDomain string
 
+	// Number of replicas for MQTT streams.
+	// Negative or 0 value means that the server(s) will pick a replica
+	// number based on the known size of the cluster (but capped at 3).
+	// Note that if an account was already connected, the stream's replica
+	// count is not modified. Use the NATS CLI to update the count if desired.
+	StreamReplicas int
+
+	// Number of replicas for MQTT consumers.
+	// Negative or 0 value means that there is no override and the consumer
+	// will have the same replica factor that the stream it belongs to.
+	// If a value is specified, it will require to be lower than the stream
+	// replicas count (lower than StreamReplicas if specified, but also lower
+	// than the automatic value determined by cluster size).
+	// Note that existing consumers are not modified.
+	ConsumerReplicas int
+
+	// Indicate if the consumers should be created with memory storage.
+	// Note that existing consumers are not modified.
+	ConsumerMemoryStorage bool
+
 	// Timeout for the authentication process.
 	AuthTimeout float64
 
@@ -4165,6 +4185,12 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 			}
 		case "js_domain":
 			o.MQTT.JsDomain = mv.(string)
+		case "stream_replicas":
+			o.MQTT.StreamReplicas = int(mv.(int64))
+		case "consumer_replicas":
+			o.MQTT.ConsumerReplicas = int(mv.(int64))
+		case "consumer_memory_storage":
+			o.MQTT.ConsumerMemoryStorage = mv.(bool)
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
diff --git a/server/reload.go b/server/reload.go
index da344c34..bce1e3f7 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -648,6 +648,33 @@ func (o *mqttMaxAckPendingReload) Apply(s *Server) {
 	s.Noticef("Reloaded: MQTT max_ack_pending = %v", o.newValue)
 }
 
+type mqttStreamReplicasReload struct {
+	noopOption
+	newValue int
+}
+
+func (o *mqttStreamReplicasReload) Apply(s *Server) {
+	s.Noticef("Reloaded: MQTT stream_replicas = %v", o.newValue)
+}
+
+type mqttConsumerReplicasReload struct {
+	noopOption
+	newValue int
+}
+
+func (o *mqttConsumerReplicasReload) Apply(s *Server) {
+	s.Noticef("Reloaded: MQTT consumer_replicas = %v", o.newValue)
+}
+
+type mqttConsumerMemoryStorageReload struct {
+	noopOption
+	newValue bool
+}
+
+func (o *mqttConsumerMemoryStorageReload) Apply(s *Server) {
+	s.Noticef("Reloaded: MQTT consumer_memory_storage = %v", o.newValue)
+}
+
 // Compares options and disconnects clients that are no longer listed in pinned certs. Lock must not be held.
 func (s *Server) recheckPinnedCerts(curOpts *Options, newOpts *Options) {
 	s.mu.Lock()
@@ -1178,12 +1205,15 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 		case "mqtt":
 			diffOpts = append(diffOpts, &mqttAckWaitReload{newValue: newValue.(MQTTOpts).AckWait})
 			diffOpts = append(diffOpts, &mqttMaxAckPendingReload{newValue: newValue.(MQTTOpts).MaxAckPending})
+			diffOpts = append(diffOpts, &mqttStreamReplicasReload{newValue: newValue.(MQTTOpts).StreamReplicas})
+			diffOpts = append(diffOpts, &mqttConsumerReplicasReload{newValue: newValue.(MQTTOpts).ConsumerReplicas})
+			diffOpts = append(diffOpts, &mqttConsumerMemoryStorageReload{newValue: newValue.(MQTTOpts).ConsumerMemoryStorage})
 			// Nil out/set to 0 the options that we allow to be reloaded so that
 			// we only fail reload if some that we don't support are changed.
 			tmpOld := oldValue.(MQTTOpts)
 			tmpNew := newValue.(MQTTOpts)
-			tmpOld.TLSConfig, tmpOld.AckWait, tmpOld.MaxAckPending = nil, 0, 0
-			tmpNew.TLSConfig, tmpNew.AckWait, tmpNew.MaxAckPending = nil, 0, 0
+			tmpOld.TLSConfig, tmpOld.AckWait, tmpOld.MaxAckPending, tmpOld.StreamReplicas, tmpOld.ConsumerReplicas, tmpOld.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
+			tmpNew.TLSConfig, tmpNew.AckWait, tmpNew.MaxAckPending, tmpNew.StreamReplicas, tmpNew.ConsumerReplicas, tmpNew.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
 			if !reflect.DeepEqual(tmpOld, tmpNew) {
 				// See TODO(ik) note below about printing old/new values.
 				return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
@@ -1191,6 +1221,9 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			}
 			tmpNew.AckWait = newValue.(MQTTOpts).AckWait
 			tmpNew.MaxAckPending = newValue.(MQTTOpts).MaxAckPending
+			tmpNew.StreamReplicas = newValue.(MQTTOpts).StreamReplicas
+			tmpNew.ConsumerReplicas = newValue.(MQTTOpts).ConsumerReplicas
+			tmpNew.ConsumerMemoryStorage = newValue.(MQTTOpts).ConsumerMemoryStorage
 		case "connecterrorreports":
 			diffOpts = append(diffOpts, &connectErrorReports{newValue: newValue.(int)})
 		case "reconnecterrorreports":


Test patch:
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 0774dcda..ba38c454 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -5533,6 +5533,354 @@ func TestMQTTClientIDInLogStatements(t *testing.T) {
 	}
 }
 
+func TestMQTTStreamReplicasOverride(t *testing.T) {
+	conf := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			stream_replicas: 3
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	cl := createJetStreamClusterWithTemplate(t, conf, "MQTT", 3)
+	defer cl.shutdown()
+
+	connectAndCheck := func(restarted bool) {
+		t.Helper()
+
+		o := cl.opts[0]
+		mc, r := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "test", cleanSess: false}, o.MQTT.Host, o.MQTT.Port, 5)
+		defer mc.Close()
+		testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, restarted)
+
+		nc, js := jsClientConnect(t, cl.servers[2])
+		defer nc.Close()
+
+		streams := []string{mqttStreamName, mqttRetainedMsgsStreamName, mqttSessStreamName}
+		for _, sn := range streams {
+			si, err := js.StreamInfo(sn)
+			require_NoError(t, err)
+			if n := len(si.Cluster.Replicas); n != 2 {
+				t.Fatalf("Expected stream %q to have 2 replicas, got %v", sn, n)
+			}
+		}
+	}
+	connectAndCheck(false)
+
+	cl.stopAll()
+	for _, o := range cl.opts {
+		o.MQTT.StreamReplicas = 2
+	}
+	cl.restartAllSamePorts()
+	cl.waitOnStreamLeader(globalAccountName, mqttStreamName)
+	cl.waitOnStreamLeader(globalAccountName, mqttRetainedMsgsStreamName)
+	cl.waitOnStreamLeader(globalAccountName, mqttSessStreamName)
+
+	l := &captureWarnLogger{warn: make(chan string, 10)}
+	cl.servers[0].SetLogger(l, false, false)
+
+	connectAndCheck(true)
+
+	select {
+	case w := <-l.warn:
+		if !strings.Contains(w, "current is 3 but configuration is 2") {
+			t.Fatalf("Unexpected warning: %q", w)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Should have warned against replicas mismatch")
+	}
+}
+
+func TestMQTTStreamReplicasConfigReload(t *testing.T) {
+	tmpl := `
+		jetstream: enabled
+		server_name: mqtt
+		mqtt {
+			port: -1
+			stream_replicas: %v
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, 3)))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	l := &captureErrorLogger{errCh: make(chan string, 10)}
+	s.SetLogger(l, false, false)
+
+	_, _, err := testMQTTConnectRetryWithError(t, &mqttConnInfo{clientID: "mqtt", cleanSess: false}, o.MQTT.Host, o.MQTT.Port, 0)
+	if err == nil {
+		t.Fatal("Expected to fail, did not")
+	}
+
+	select {
+	case e := <-l.errCh:
+		if !strings.Contains(e, NewJSStreamReplicasNotSupportedError().Description) {
+			t.Fatalf("Expected error regarding replicas, got %v", e)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("Did not get the error regarding replicas count")
+	}
+
+	reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, 1))
+
+	mc, r := testMQTTConnect(t, &mqttConnInfo{clientID: "mqtt", cleanSess: false}, o.MQTT.Host, o.MQTT.Port)
+	defer mc.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+}
+
+func TestMQTTStreamReplicasInsufficientResources(t *testing.T) {
+	conf := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			stream_replicas: 5
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	cl := createJetStreamClusterWithTemplate(t, conf, "MQTT", 3)
+	defer cl.shutdown()
+
+	l := &captureErrorLogger{errCh: make(chan string, 10)}
+	for _, s := range cl.servers {
+		s.SetLogger(l, false, false)
+	}
+
+	o := cl.opts[1]
+	_, _, err := testMQTTConnectRetryWithError(t, &mqttConnInfo{clientID: "mqtt", cleanSess: false}, o.MQTT.Host, o.MQTT.Port, 0)
+	if err == nil {
+		t.Fatal("Expected to fail, did not")
+	}
+
+	select {
+	case e := <-l.errCh:
+		if !strings.Contains(e, NewJSInsufficientResourcesError().Description) {
+			t.Fatalf("Expected error regarding insufficient resources, got %v", e)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("Did not get the error regarding replicas count")
+	}
+}
+
+func TestMQTTConsumerReplicasValidate(t *testing.T) {
+	o := testMQTTDefaultOptions()
+	for _, test := range []struct {
+		name string
+		sr   int
+		cr   int
+		err  bool
+	}{
+		{"stream replicas neg", -1, 3, false},
+		{"stream replicas 0", 0, 3, false},
+		{"consumer replicas neg", 0, -1, false},
+		{"consumer replicas 0", -1, 0, false},
+		{"consumer replicas too high", 1, 2, true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			o.MQTT.StreamReplicas = test.sr
+			o.MQTT.ConsumerReplicas = test.cr
+			err := validateMQTTOptions(o)
+			if test.err {
+				if err == nil {
+					t.Fatal("Expected error, did not get one")
+				}
+				if !strings.Contains(err.Error(), "cannot be higher") {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				// OK
+				return
+			} else if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+		})
+	}
+}
+
+func TestMQTTConsumerReplicasOverride(t *testing.T) {
+	conf := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			stream_replicas: 5
+			consumer_replicas: 1
+			consumer_memory_storage: true
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	cl := createJetStreamClusterWithTemplate(t, conf, "MQTT", 5)
+	defer cl.shutdown()
+
+	connectAndCheck := func(subject string, restarted bool) {
+		t.Helper()
+
+		o := cl.opts[0]
+		mc, r := testMQTTConnect(t, &mqttConnInfo{clientID: "test", cleanSess: false}, o.MQTT.Host, o.MQTT.Port)
+		defer mc.Close()
+		testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, restarted)
+		testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+
+		nc, js := jsClientConnect(t, cl.servers[2])
+		defer nc.Close()
+
+		for ci := range js.ConsumersInfo(mqttStreamName) {
+			if ci.Config.FilterSubject == mqttStreamSubjectPrefix+"foo" {
+				if len(ci.Cluster.Replicas) != 0 {
+					t.Fatalf("Expected consumer to be R1, got: %+v", ci.Cluster)
+				}
+			} else {
+				if len(ci.Cluster.Replicas) != 1 {
+					t.Fatalf("Expected consumer to be R2, got: %+v", ci.Cluster)
+				}
+			}
+		}
+	}
+	connectAndCheck("foo", false)
+
+	cl.stopAll()
+	for _, o := range cl.opts {
+		o.MQTT.ConsumerReplicas = 2
+		o.MQTT.ConsumerMemoryStorage = false
+	}
+	cl.restartAllSamePorts()
+	cl.waitOnStreamLeader(globalAccountName, mqttStreamName)
+	cl.waitOnStreamLeader(globalAccountName, mqttRetainedMsgsStreamName)
+	cl.waitOnStreamLeader(globalAccountName, mqttSessStreamName)
+
+	connectAndCheck("bar", true)
+}
+
+func TestMQTTConsumerReplicasReload(t *testing.T) {
+	tmpl := `
+		jetstream: enabled
+		server_name: mqtt
+		mqtt {
+			port: -1
+			consumer_replicas: %v
+			consumer_memory_storage: %s
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, 3, "false")))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	l := &captureErrorLogger{errCh: make(chan string, 10)}
+	s.SetLogger(l, false, false)
+
+	c, r := testMQTTConnect(t, &mqttConnInfo{clientID: "sub", cleanSess: false}, o.MQTT.Host, o.MQTT.Port)
+	defer c.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{mqttSubAckFailure})
+
+	select {
+	case e := <-l.errCh:
+		if !strings.Contains(e, NewJSStreamReplicasNotSupportedError().Description) {
+			t.Fatalf("Expected error regarding replicas, got %v", e)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("Did not get the error regarding replicas count")
+	}
+
+	reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, 1, "true"))
+
+	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+
+	mset, err := s.GlobalAccount().lookupStream(mqttStreamName)
+	if err != nil {
+		t.Fatalf("Error looking up stream: %v", err)
+	}
+	var cons *consumer
+	mset.mu.RLock()
+	for _, c := range mset.consumers {
+		cons = c
+		break
+	}
+	mset.mu.RUnlock()
+	cons.mu.RLock()
+	st := cons.store.Type()
+	cons.mu.RUnlock()
+	if st != MemoryStorage {
+		t.Fatalf("Expected storage %v, got %v", MemoryStorage, st)
+	}
+}
+
+func TestMQTTConsumerReplicasExceedsParentStream(t *testing.T) {
+	conf := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			consumer_replicas: 4
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	cl := createJetStreamClusterWithTemplate(t, conf, "MQTT", 3)
+	defer cl.shutdown()
+
+	l := &captureErrorLogger{errCh: make(chan string, 10)}
+	for _, s := range cl.servers {
+		s.SetLogger(l, false, false)
+	}
+
+	o := cl.opts[0]
+	mc, r := testMQTTConnect(t, &mqttConnInfo{clientID: "test", cleanSess: false}, o.MQTT.Host, o.MQTT.Port)
+	defer mc.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{mqttSubAckFailure})
+
+	select {
+	case e := <-l.errCh:
+		if !strings.Contains(e, NewJSConsumerReplicasExceedsStreamError().Description) {
+			t.Fatalf("Expected error regarding replicas exceeded parent, got %v", e)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("Did not get the error regarding replicas count")
+	}
+}
+
 //////////////////////////////////////////////////////////////////////////
 //
 // Benchmarks