{"record":{"id":"70f6487e80587f6e","repo":"nats-io/nats-server","slug":"consumer-closed","errorCode":null,"errorMessage":"consumer closed","messagePattern":"consumer closed","errorType":"exception","errorClass":"errConsumerClosed","httpStatus":null,"severity":"warning","filePath":"server/jetstream_cluster.go","lineNumber":8046,"sourceCode":"\t\t\t\t}\n\t\t\t\to.mu.Unlock()\n\t\t\tcase removePendingRequest:\n\t\t\t\to.mu.Lock()\n\t\t\t\tif !o.isLeader() {\n\t\t\t\t\tif o.prm != nil {\n\t\t\t\t\t\tdelete(o.prm, string(buf[1:]))\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t\to.mu.Unlock()\n\t\t\tdefault:\n\t\t\t\treturn fmt.Errorf(\"unknown consumer entry op type: %v\", entryOp(buf[0]))\n\t\t\t}\n\t\t}\n\t}\n\treturn nil\n}\n\nvar errConsumerClosed = errors.New(\"consumer closed\")\n\nfunc (o *consumer) processReplicatedAck(dseq, sseq uint64) error {\n\to.mu.Lock()\n\t// Update activity.\n\to.lat = time.Now()\n\n\tvar ackAllSeqs []uint64\n\tif o.retention != LimitsPolicy && (o.cfg.AckPolicy == AckAll || o.cfg.AckPolicy == AckFlowControl) {\n\t\t// Always use the store state, as o.asflr is skipped ahead already.\n\t\t// Capture before updating store, which clears the pending below.\n\t\tstate, err := o.store.BorrowState()\n\t\tif err == nil {\n\t\t\t// Only need to collect if the ack covers more than the sequence itself.\n\t\t\tif sagap := sseq - state.AckFloor.Stream; sagap > 1 {\n\t\t\t\t// At most the pending entries below the ack, don't over-allocate.\n\t\t\t\tackAllSeqs = make([]uint64, 0, min(uint64(len(state.Pending)), sagap-1))\n\t\t\t\tfor seq := range state.Pending {\n\t\t\t\t\tif seq < sseq {","sourceCodeStart":8028,"sourceCodeEnd":8064,"githubUrl":"https://github.com/nats-io/nats-server/blob/3a66a489d262bf89b71a71c955c94920394532f3/server/jetstream_cluster.go#L8028-L8064","documentation":"`errConsumerClosed` is a sentinel error from the JetStream consumer code: it signals that a replicated consumer's `Consumer` object has been closed (e.g. due to leader change, consumer deletion, or shutdown) while its raft group was still being used. Callers deliberately special-case it (server/jetstream_cluster.go:7567, 7950) to avoid logging spurious errors during normal leader transitions. Exported paths that surface it indicate the consumer ceased to exist mid-operation.","triggerScenarios":"`o.processReplicatedAck(dseq, sseq)` on an already-closed consumer; applying replicated consumer entries to a consumer whose raft group was torn down; snapshot/entry application racing a consumer delete or leadership loss (server/jetstream_cluster.go:8046).","commonSituations":"Deleting a pull consumer while clients are still acking messages; frequent leader elections in a JetStream cluster causing consumer reassignment; server shutdown or asset rebalancing racing consumer operations.","solutions":["Confirm the consumer still exists with `nats consumer info <stream> <consumer>`; recreate it if it was deleted","Retry the operation after the new consumer leader stabilizes — the error during leader change is expected and self-healing","Stop client ack/subscribe loops gracefully before deleting consumers","Check cluster stability (flapping leader elections) if the error recurs frequently"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Verify the consumer still exists before acking/subscribing\n_, err := js.ConsumerInfo(stream, consumer)\nif err != nil { recreateOrLookupConsumer() }","typeGuard":null,"tryCatchPattern":"// Treat 'consumer closed' as transient during leader elections\nif err := o.processReplicatedAck(dseq, sseq); err != nil {\n  if errors.Is(err, errConsumerClosed) { /* resync with new leader, do not log fatal */ }\n}","preventionTips":["Stop consumers gracefully before deleting them","Stabilize the JetStream cluster to avoid frequent leader elections","Recreate consumers with durable names so clients can resubscribe after closure"],"tags":["jetstream","consumer","cluster","leader-election"],"backgroundTag":"consumer-closed","analyzedSha":"3a66a489d262bf89b71a71c955c94920394532f3","analyzedAt":"2026-09-02T04:41:54.247Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}