Repository: nats-io/nats-server
Pull request #3817 (2023-01-26 01:18:33)

[FIXED] Force consumer replicas to match for interest policy streams
Consumers must match replica of parent stream if interest based policy retention.

For MQTT, this was ill-advised by me, not understanding that the messages stream for MQTT was interest policy based. Interest policy based streams require consumers to match the replica count.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index b5d5adc2..66963c1f 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -382,6 +382,16 @@ func checkConsumerCfg(
 	if config.Replicas < 0 {
 		return NewJSReplicasCountCannotBeNegativeError()
 	}
+	// If the stream is interest or workqueue retention make sure the replicas
+	// match that of the stream. This is REQUIRED for now.
+	if cfg.Retention == InterestPolicy || cfg.Retention == WorkQueuePolicy {
+		// Only error here if not recovering.
+		// We handle recovering in a different spot to allow consumer to come up
+		// if previous version allowed it to be created. We do not want it to not come up.
+		if !isRecovering && config.Replicas != 0 && config.Replicas != cfg.Replicas {
+			return NewJSConsumerReplicasShouldMatchStreamError()
+		}
+	}
 
 	// Check if we have a BackOff defined that MaxDeliver is within range etc.
 	if lbo := len(config.BackOff); lbo > 0 && config.MaxDeliver <= lbo {
diff --git a/server/errors.json b/server/errors.json
index 2413531b..e48eea78 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1318,5 +1318,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerReplicasShouldMatchStream",
+    "code": 400,
+    "error_code": 10134,
+    "description": "consumer config replicas must match interest retention stream's replicas",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/jetstream.go b/server/jetstream.go
index 63e7dadc..8ca10d47 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1381,6 +1381,7 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 	// Check interest policy streams for auto cleanup.
 	for _, mset := range ipstreams {
 		mset.checkForOrphanMsgs()
+		mset.checkConsumerReplication()
 	}
 
 	s.Debugf("JetStream state for account %q recovered", a.Name)
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 8cf409c3..58bdfe30 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -167,6 +167,9 @@ const (
 	// JSConsumerReplicasExceedsStream consumer config replica count exceeds parent stream
 	JSConsumerReplicasExceedsStream ErrorIdentifier = 10126
 
+	// JSConsumerReplicasShouldMatchStream consumer config replicas must match interest retention stream's replicas
+	JSConsumerReplicasShouldMatchStream ErrorIdentifier = 10134
+
 	// JSConsumerSmallHeartbeatErr consumer idle heartbeat needs to be >= 100ms
 	JSConsumerSmallHeartbeatErr ErrorIdentifier = 10083
 
@@ -458,6 +461,7 @@ var (
 		JSConsumerPushMaxWaitingErr:                {Code: 400, ErrCode: 10080, Description: "consumer in push mode can not set max waiting"},
 		JSConsumerReplacementWithDifferentNameErr:  {Code: 400, ErrCode: 10106, Description: "consumer replacement durable config not the same"},
 		JSConsumerReplicasExceedsStream:            {Code: 400, ErrCode: 10126, Description: "consumer config replica count exceeds parent stream"},
+		JSConsumerReplicasShouldMatchStream:        {Code: 400, ErrCode: 10134, Description: "consumer config replicas must match interest retention stream's replicas"},
 		JSConsumerSmallHeartbeatErr:                {Code: 400, ErrCode: 10083, Description: "consumer idle heartbeat needs to be >= 100ms"},
 		JSConsumerStoreFailedErrF:                  {Code: 500, ErrCode: 10104, Description: "error creating store for consumer: {err}"},
 		JSConsumerWQConsumerNotDeliverAllErr:       {Code: 400, ErrCode: 10101, Description: "consumer must be deliver all on workqueue stream"},
@@ -1149,6 +1153,16 @@ func NewJSConsumerReplicasExceedsStreamError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSConsumerReplicasExceedsStream]
 }
 
+// NewJSConsumerReplicasShouldMatchStreamError creates a new JSConsumerReplicasShouldMatchStream error: "consumer config replicas must match interest retention stream's replicas"
+func NewJSConsumerReplicasShouldMatchStreamError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSConsumerReplicasShouldMatchStream]
+}
+
 // NewJSConsumerSmallHeartbeatError creates a new JSConsumerSmallHeartbeatErr error: "consumer idle heartbeat needs to be >= 100ms"
 func NewJSConsumerSmallHeartbeatError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/mqtt.go b/server/mqtt.go
index 949162b1..054523da 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -3701,9 +3701,6 @@ func (sess *mqttSession) processJSConsumer(c *client, subject, sid string,
 			MaxAckPending:  maxAckPending,
 			MemoryStorage:  opts.MQTT.ConsumerMemoryStorage,
 		}
-		if r := opts.MQTT.ConsumerReplicas; r > 0 {
-			cc.Replicas = r
-		}
 		if opts.MQTT.ConsumerInactiveThreshold > 0 {
 			cc.InactiveThreshold = opts.MQTT.ConsumerInactiveThreshold
 		}
diff --git a/server/opts.go b/server/opts.go
index 8bf29a79..dfd12165 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -439,6 +439,9 @@ type MQTTOpts struct {
 	// 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.
+	//
+	// UPDATE: This is no longer used while messages stream has interest policy retention
+	// which requires consumer replica count to match the parent stream.
 	ConsumerReplicas int
 
 	// Indicate if the consumers should be created with memory storage.
@@ -4223,7 +4226,14 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 		case "stream_replicas":
 			o.MQTT.StreamReplicas = int(mv.(int64))
 		case "consumer_replicas":
-			o.MQTT.ConsumerReplicas = int(mv.(int64))
+			err := &configWarningErr{
+				field: mk,
+				configErr: configErr{
+					token:  tk,
+					reason: `consumer replicas setting ignored in this server version`,
+				},
+			}
+			*warnings = append(*warnings, err)
 		case "consumer_memory_storage":
 			o.MQTT.ConsumerMemoryStorage = mv.(bool)
 		case "consumer_inactive_threshold", "consumer_auto_cleanup":
diff --git a/server/stream.go b/server/stream.go
index 0c1cd095..bfaf3fd6 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -5028,3 +5028,24 @@ func (mset *stream) checkForOrphanMsgs() {
 		}
 	}
 }
+
+// Check on startup to make sure that consumers replication matches us.
+// Interest retention requires replication matches.
+func (mset *stream) checkConsumerReplication() {
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+
+	if mset.cfg.Retention != InterestPolicy {
+		return
+	}
+
+	s, acc := mset.srv, mset.acc
+	for _, o := range mset.consumers {
+		o.mu.RLock()
+		if mset.cfg.Replicas != o.cfg.Replicas {
+			s.Errorf("consumer '%s > %s > %s' MUST match replication (%d vs %d) of stream with interest policy",
+				acc, mset.cfg.Name, o.cfg.Name, mset.cfg.Replicas, o.cfg.Replicas)
+		}
+		o.mu.RUnlock()
+	}
+}


Test patch:
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index ce865116..30017af1 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -2463,6 +2463,34 @@ func TestJetStreamClusterDirectGetStreamUpgrade(t *testing.T) {
 	require_True(t, string(entry.Value()) == "derek")
 }
 
+// For interest (or workqueue) based streams its important to match the replication factor.
+// This was the case but now that more control over consumer creation is allowed its possible
+// to create a consumer where the replication factor does not match. This could cause
+// instability in the state between servers and cause problems on leader switches.
+func TestJetStreamClusterInterestPolicyStreamForConsumersToMatchRFactor(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Subjects:  []string{"foo"},
+		Retention: nats.InterestPolicy,
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "XX",
+		AckPolicy: nats.AckExplicitPolicy,
+		Replicas:  1,
+	})
+
+	require_Error(t, err, NewJSConsumerReplicasShouldMatchStreamError())
+}
+
 // https://github.com/nats-io/nats-server/issues/3791
 func TestJetStreamClusterKVWatchersWithServerDown(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index f734652d..b5dd4bbc 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -5788,6 +5788,9 @@ func TestMQTTConsumerReplicasValidate(t *testing.T) {
 	}
 }
 
+// This was ill-advised since the messages stream is currently interest policy.
+// Interest policy streams require consumers match the replica count.
+// Will leave her for now to make sure we do not override.
 func TestMQTTConsumerReplicasOverride(t *testing.T) {
 	conf := `
 		listen: 127.0.0.1:-1
@@ -5820,19 +5823,15 @@ func TestMQTTConsumerReplicasOverride(t *testing.T) {
 		mc, r := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "test", cleanSess: false}, o.MQTT.Host, o.MQTT.Port, 5)
 		defer mc.Close()
 		testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, restarted)
-		testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+		testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: subject, 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)
+			if ci.Config.FilterSubject == mqttStreamSubjectPrefix+subject {
+				if rf := len(ci.Cluster.Replicas) + 1; rf != 5 {
+					t.Fatalf("Expected consumer to be R5, got: %d", rf)
 				}
 			}
 		}
@@ -5852,17 +5851,16 @@ func TestMQTTConsumerReplicasOverride(t *testing.T) {
 	connectAndCheck("bar", true)
 }
 
-func TestMQTTConsumerReplicasReload(t *testing.T) {
+func TestMQTTConsumerMemStorageReload(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")))
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, "false")))
 	s, o := RunServerWithConfig(conf)
 	defer testMQTTShutdownServer(s)
 
@@ -5872,18 +5870,8 @@ func TestMQTTConsumerReplicasReload(t *testing.T) {
 	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"))
+	reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, "true"))
 
 	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
 
@@ -5906,50 +5894,6 @@ func TestMQTTConsumerReplicasReload(t *testing.T) {
 	}
 }
 
-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 := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "test", cleanSess: false}, o.MQTT.Host, o.MQTT.Port, 2)
-	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")
-	}
-}
-
 type unableToDeleteConsLogger struct {
 	DummyLogger
 	errCh chan string