micro/go-micro · error

Error subscribing to topic

Error message

Error subscribing to topic

What it means

After ensuring the stream exists, Consume subscribes either via QueueSubscribe (durable streams enabled) or Subscribe, attaching the message handler and subscription options. This error wraps any NATS failure returned from creating that subscription.

Source

Thrown at events/natsjs/nats.go:270

	} else {
		subOpts = append(subOpts, nats.DeliverNew())
	}

	if options.AckWait > 0 {
		subOpts = append(subOpts, nats.AckWait(options.AckWait))
	}

	// connect the subscriber via a queue group only if durable streams are enabled
	if !s.opts.DisableDurableStreams {
		subOpts = append(subOpts, nats.Durable(options.Group))
		_, err = s.natsJetStreamCtx.QueueSubscribe(topic, options.Group, handleMsg, subOpts...)
	} else {
		subOpts = append(subOpts, nats.ConsumerName(options.Group))
		_, err = s.natsJetStreamCtx.Subscribe(topic, handleMsg, subOpts...)
	}

	if err != nil {
		return nil, errors.Wrap(err, "Error subscribing to topic")
	}

	return channel, nil
}

// Close implements io.Closer and closes the underlying NATS connection.
// This method is optional but recommended to prevent connection leaks.
func (s *stream) Close() error {
	if s.conn != nil {
		s.conn.Close()
		s.conn = nil
	}
	return nil
}

// Ensure stream implements io.Closer
var _ io.Closer = (*stream)(nil)

View on GitHub (pinned to 24529f1404)

Solutions

1. Read errors.Cause(err) to see the NATS subscription failure (e.g. consumer name already in use)
2. Delete the stale durable consumer (nats consumer rm or js.DeleteConsumer) when its config changed
3. Keep Consume options (Group, AutoAck, AckWait, retries) stable across deployments, or change Group name
4. Ensure the NATS connection is healthy before calling Consume
5. Verify the stream for the topic still exists at subscribe time

Example fix

// before
// changed AckWait but old durable "order-svc" exists -> consumer name already in use
_, err = js.QueueSubscribe(topic, "order-svc", handleMsg, subOpts...)
// after
js.DeleteConsumer(streamName, "order-svc") // remove stale durable once before resubscribing
_, err = js.QueueSubscribe(topic, "order-svc", handleMsg, subOpts...)

// ensure connection and stream are healthy before subscribing
if !conn.IsConnected() { return errors.New("nats not connected") }
if _, err := js.StreamInfo(topic); err != nil { return err }

ch, err := stream.Consume(topic, events.Group("order-svc"))
if err != nil {
    if strings.Contains(err.Error(), "in use") || strings.Contains(err.Error(), "exists") {
        // delete stale durable and retry once
        js.DeleteConsumer(topic, "order-svc")
        ch, err = stream.Consume(topic, events.Group("order-svc"))
    }
    if err != nil { return err }
}

When it happens

Trigger: Calling Consume when the subscription fails: a durable consumer with the same name already exists with a different configuration, the topic/stream was deleted between the StreamInfo check and Subscribe, or the connection dropped before the subscribe completed.

Common situations: Changing ConsumeOptions (AckWait, MaxDeliver, start time) while a durable consumer of the same name exists — NATS rejects reconfiguring durables; restarting a service against a stale durable consumer; NATS connection loss during startup.

Related errors

• source not found: %s
• error connecting to nats cluster %v: %w
• error while obtaining JetStream context: %w
• Stream did not exist and adding a stream failed
• Failed to connect to NATS Server

AI-assisted analysis of micro/go-micro@24529f1404 (2026-09-01). Data as JSON: /api/errors/4aca5301ca71760e. Report an issue: GitHub.