Repository: nats-io/nats-server
Pull request #3193 (2022-06-15 00:53:44)

Allow for MQTT QoS-1 consumers to be auto cleaned up
A server can configure the mqtt section to designate orphaned inactive consumers for auto cleanup.
Several users have reported a buildup of orphaned consumers from MQTT software they can not change or update.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/mqtt.go b/server/mqtt.go
index 30f23b62..1c903b56 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -3622,6 +3622,9 @@ func (sess *mqttSession) processJSConsumer(c *client, subject, sid string,
 		if r := opts.MQTT.ConsumerReplicas; r > 0 {
 			cc.Replicas = r
 		}
+		if opts.MQTT.ConsumerInactiveThreshold > 0 {
+			cc.InactiveThreshold = opts.MQTT.ConsumerInactiveThreshold
+		}
 		if err := sess.createConsumer(cc); err != nil {
 			c.Errorf("Unable to add JetStream consumer for subscription on %q: err=%v", subject, err)
 			return nil, nil, err
diff --git a/server/opts.go b/server/opts.go
index 3a32d5ea..db5a1a56 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -431,6 +431,10 @@ type MQTTOpts struct {
 	// Note that existing consumers are not modified.
 	ConsumerMemoryStorage bool
 
+	// If specified will have the system auto-cleanup the consumers after being
+	// inactive for the specified amount of time.
+	ConsumerInactiveThreshold time.Duration
+
 	// Timeout for the authentication process.
 	AuthTimeout float64
 
@@ -4191,6 +4195,9 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 			o.MQTT.ConsumerReplicas = int(mv.(int64))
 		case "consumer_memory_storage":
 			o.MQTT.ConsumerMemoryStorage = mv.(bool)
+		case "consumer_inactive_threshold", "consumer_auto_cleanup":
+			o.MQTT.ConsumerInactiveThreshold = parseDuration("consumer_inactive_threshold", tk, mv, errors, warnings)
+
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
diff --git a/server/reload.go b/server/reload.go
index bce1e3f7..38658d5e 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -675,6 +675,15 @@ func (o *mqttConsumerMemoryStorageReload) Apply(s *Server) {
 	s.Noticef("Reloaded: MQTT consumer_memory_storage = %v", o.newValue)
 }
 
+type mqttInactiveThresholdReload struct {
+	noopOption
+	newValue time.Duration
+}
+
+func (o *mqttInactiveThresholdReload) Apply(s *Server) {
+	s.Noticef("Reloaded: MQTT consumer_inactive_threshold = %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()
@@ -1208,12 +1217,17 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			diffOpts = append(diffOpts, &mqttStreamReplicasReload{newValue: newValue.(MQTTOpts).StreamReplicas})
 			diffOpts = append(diffOpts, &mqttConsumerReplicasReload{newValue: newValue.(MQTTOpts).ConsumerReplicas})
 			diffOpts = append(diffOpts, &mqttConsumerMemoryStorageReload{newValue: newValue.(MQTTOpts).ConsumerMemoryStorage})
+			diffOpts = append(diffOpts, &mqttInactiveThresholdReload{newValue: newValue.(MQTTOpts).ConsumerInactiveThreshold})
+
 			// 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, tmpOld.StreamReplicas, tmpOld.ConsumerReplicas, tmpOld.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
+			tmpOld.ConsumerInactiveThreshold = 0
 			tmpNew.TLSConfig, tmpNew.AckWait, tmpNew.MaxAckPending, tmpNew.StreamReplicas, tmpNew.ConsumerReplicas, tmpNew.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
+			tmpNew.ConsumerInactiveThreshold = 0
+
 			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",
@@ -1224,6 +1238,7 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			tmpNew.StreamReplicas = newValue.(MQTTOpts).StreamReplicas
 			tmpNew.ConsumerReplicas = newValue.(MQTTOpts).ConsumerReplicas
 			tmpNew.ConsumerMemoryStorage = newValue.(MQTTOpts).ConsumerMemoryStorage
+			tmpNew.ConsumerInactiveThreshold = newValue.(MQTTOpts).ConsumerInactiveThreshold
 		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 465abcef..3913844a 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -6001,6 +6001,54 @@ func TestMQTTSessionNotDeletedOnDeleteConsumerError(t *testing.T) {
 	require_True(t, si.State.Msgs == 1)
 }
 
+// Test for auto-cleanup of consumers.
+func TestMQTTConsumerInactiveThreshold(t *testing.T) {
+	tmpl := `
+		listen: 127.0.0.1:-1
+		server_name: mqtt
+		jetstream: enabled
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			consumer_inactive_threshold: %q
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, "0.2s")))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	mc, r := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "test", cleanSess: true}, o.MQTT.Host, o.MQTT.Port, 5)
+	defer mc.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	ci := <-js.ConsumersInfo("$MQTT_msgs")
+
+	// Make sure we clean up this consumer.
+	mc.Close()
+	checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
+		_, err := js.ConsumerInfo(ci.Stream, ci.Name)
+		if err == nil {
+			return fmt.Errorf("Consumer still present")
+		}
+		return nil
+	})
+
+	// Check reload.
+	// We will not redo existing consumers however.
+	reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, "22s"))
+	if opts := s.getOpts(); opts.MQTT.ConsumerInactiveThreshold != 22*time.Second {
+		t.Fatalf("Expected reloaded value of %v but got %v", 22*time.Second, opts.MQTT.ConsumerInactiveThreshold)
+	}
+}
+
 //////////////////////////////////////////////////////////////////////////
 //
 // Benchmarks