{"record":{"id":"07be5eb81b465b90","repo":"nats-io/nats-server","slug":"create-retained-messages-consumer-for-account-q","errorCode":null,"errorMessage":"create retained messages consumer for account %q: %v","messagePattern":"create retained messages consumer for account %q: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"server/mqtt.go","lineNumber":1604,"sourceCode":"\t// before. Ignore any errors that might arise.\n\trmLegacyDurName := mqttRetainedMsgsStreamName + \"_\" + jsa.id\n\tjsa.deleteConsumer(mqttRetainedMsgsStreamName, rmLegacyDurName, true)\n\n\t// Create a new, uniquely names consumer for retained messages for this\n\t// server. The prior one will expire eventually.\n\tccfg := &CreateConsumerRequest{\n\t\tStream: mqttRetainedMsgsStreamName,\n\t\tConfig: ConsumerConfig{\n\t\t\tName:              mqttRetainedMsgsStreamName + \"_\" + nuid.Next(),\n\t\t\tFilterSubject:     mqttRetainedMsgsStreamSubject + \">\",\n\t\t\tDeliverSubject:    rmsubj,\n\t\t\tReplayPolicy:      ReplayInstant,\n\t\t\tAckPolicy:         AckNone,\n\t\t\tInactiveThreshold: 5 * time.Minute,\n\t\t},\n\t}\n\tif _, err := jsa.createEphemeralConsumer(ccfg); err != nil {\n\t\treturn nil, fmt.Errorf(\"create retained messages consumer for account %q: %v\", accName, err)\n\t}\n\n\t// Set this so that on defer we don't cleanup.\n\tsuccess = true\n\n\treturn as, nil\n}\n\nfunc (s *Server) mqttDetermineReplicas() int {\n\t// If not clustered, then replica will be 1.\n\tif !s.JetStreamIsClustered() {\n\t\treturn 1\n\t}\n\topts := s.getOpts()\n\treplicas := 0\n\tfor _, u := range opts.Routes {\n\t\thost := u.Hostname()\n\t\t// If this is an IP just add one.","sourceCodeStart":1586,"sourceCodeEnd":1622,"githubUrl":"https://github.com/nats-io/nats-server/blob/3a66a489d262bf89b71a71c955c94920394532f3/server/mqtt.go#L1586-L1622","documentation":"The server creates an ephemeral JetStream consumer to drain MQTT retained messages during session/account setup; if createEphemeralConsumer fails the error is wrapped with the account name. This means the internal consumer for the retained messages stream could not be created, so MQTT retained message handling cannot proceed.","triggerScenarios":"Retained messages stream exists but is in a bad state (no responders/leadership); JetStream API request for consumer creation times out; consumer limits reached on the account/stream; stream unavailable mid-creation.","commonSituations":"Cluster without a healthy JS leader for the stream; account consumer limits exceeded by leaking ephemeral consumers; transient network/raft issues during MQTT connect.","solutions":["Check consumer limits on the account/stream and prune stale ephemeral consumers","Verify JetStream cluster health (stream and consumer leaders elected) via nats server report jetstream","Retry the MQTT connection once JS is healthy","Inspect logs for the wrapped cause (e.g. 'no responders', 'leader election in progress') and address underlying JS issue"],"exampleFix":"// before\nnats consumer report $MQTT.rmsgs   # many stale ephemeral consumers\n// after\nnats consumer rm $MQTT.rmsgs <stale-ephemeral-consumer>\n# then reconnect the MQTT client","handlingStrategy":"retry","validationCode":"// Check the stream is ready to accept consumers:\njs, _ := nc.JetStream()\nsi, err := js.StreamInfo(\"$MQTT.rmsgs\")\nif err == nil && len(si.State.Replicas) > 0 {\n    // verify no current leader election in progress\n}","typeGuard":null,"tryCatchPattern":"if strings.Contains(err.Error(), \"create retained messages consumer\") {\n    // often transient (elections/no responders); retry with backoff\n    return retryWithBackoff(reconnectMQTT)\n}","preventionTips":["Set account consumer limits high enough for MQTT sessions (one ephemeral per session)","Alert on JS leader elections; pause mass MQTT reconnects during them","Prune leaked ephemeral consumers regularly"],"tags":["jetstream","mqtt","consumer","ephemeral-consumer"],"backgroundTag":"jetstream-consumer-create-failed","analyzedSha":"3a66a489d262bf89b71a71c955c94920394532f3","analyzedAt":"2026-09-02T04:41:54.247Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}