[CHANGED] MQTT RMS consumer: instead of deleting/creating, create a u…

server/mqtt.go

@@ -1439,26 +1439,32 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 		}
 	}
 
+	// Delete the old (legacy) consumer, from v2.10.10 and before.
+	rmLegacyDurName := mqttRetainedMsgsStreamName + "_" + jsa.id
+	if _, err := jsa.deleteConsumer(mqttRetainedMsgsStreamName, rmLegacyDurName); isErrorOtherThan(err, JSConsumerNotFoundErr) {
+		return nil, err
+	}
+
 	// 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 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
-	// If error other than "not found" then fail, otherwise proceed with creating
-	// the durable consumer.
-	if _, err := jsa.deleteConsumer(mqttRetainedMsgsStreamName, rmDurName); isErrorOtherThan(err, JSConsumerNotFoundErr) {
-		return nil, err
-	}
+	// retained messages.
+	//
+	// So we use a durable consumer, and create a new one each time we start.
+	// The old one should expire and get deleted due to inactivity. The name for
+	// the durable is $MQTT_rmsgs_{uuid}_{server-name}, the server name is just
+	// for readability.
+	rmDurName := mqttRetainedMsgsStreamName + "_" + nuid.Next() + "_" + s.String()
+
 	ccfg := &CreateConsumerRequest{
 		Stream: mqttRetainedMsgsStreamName,
 		Config: ConsumerConfig{
-			Durable:        rmDurName,
-			FilterSubject:  mqttRetainedMsgsStreamSubject + ">",
-			DeliverSubject: rmsubj,
-			ReplayPolicy:   ReplayInstant,
-			AckPolicy:      AckNone,
+			Durable:           rmDurName,
+			FilterSubject:     mqttRetainedMsgsStreamSubject + ">",
+			DeliverSubject:    rmsubj,
+			ReplayPolicy:      ReplayInstant,
+			AckPolicy:         AckNone,
+			InactiveThreshold: 5 * time.Minute,
 		},
 	}
 	if _, err := jsa.createConsumer(ccfg); err != nil {

server/mqtt_test.go

@@ -3026,8 +3026,10 @@ func TestMQTTCluster(t *testing.T) {
 				for _, sn := range streams {
 					cl.waitOnStreamLeader(globalAccountName, sn)
 				}
-				cl.waitOnConsumerLeader(globalAccountName, mqttRetainedMsgsStreamName, "$MQTT_rmsgs_esFhDys3")
-				cl.waitOnConsumerLeader(globalAccountName, mqttRetainedMsgsStreamName, "$MQTT_rmsgs_z3WIzPtj")
+				// <>/<> TODO: need to find a way to wait for the consumer
+				// leader, but the names are unique and not predictable.
+				// cl.waitOnConsumerLeader(globalAccountName, mqttRetainedMsgsStreamName, "$MQTT_rmsgs_esFhDys3")
+				// cl.waitOnConsumerLeader(globalAccountName, mqttRetainedMsgsStreamName, "$MQTT_rmsgs_z3WIzPtj")
 			}
 		})
 	}