Repository: nats-io/nats.go
Pull request #1401 (2023-09-14 08:38:09)

[ADDED] Customizing queue groups in micro
Addresses ADR-32 changes from https://github.com/nats-io/nats-architecture-and-design/pull/239


Patch:
diff --git a/jetstream/options.go b/jetstream/options.go
index e1b4b5b..566f0dc 100644
--- a/jetstream/options.go
+++ b/jetstream/options.go
@@ -164,6 +164,11 @@ func (t PullThresholdMessages) configureConsume(opts *consumeOpts) error {
 	return nil
 }
 
+func (t PullThresholdMessages) configureMessages(opts *consumeOpts) error {
+	opts.ThresholdMessages = int(t)
+	return nil
+}
+
 // PullThresholdBytes sets the byte count on which Consume will trigger
 // new pull request to the server. Defaults to 50% of MaxBytes (if set).
 type PullThresholdBytes int
@@ -173,6 +178,11 @@ func (t PullThresholdBytes) configureConsume(opts *consumeOpts) error {
 	return nil
 }
 
+func (t PullThresholdBytes) configureMessages(opts *consumeOpts) error {
+	opts.ThresholdBytes = int(t)
+	return nil
+}
+
 // PullHeartbeat sets the idle heartbeat duration for a pull subscription
 // If a client does not receive a heartbeat message from a stream for more
 // than the idle heartbeat setting, the subscription will be removed
@@ -197,6 +207,25 @@ func (hb PullHeartbeat) configureMessages(opts *consumeOpts) error {
 	return nil
 }
 
+// StopAfter sets the number of messages after which the consumer is automatically stopped
+type StopAfter int
+
+func (nMsgs StopAfter) configureConsume(opts *consumeOpts) error {
+	if nMsgs <= 0 {
+		return fmt.Errorf("%w: auto stop after value cannot be less than 1", ErrInvalidOption)
+	}
+	opts.StopAfter = int(nMsgs)
+	return nil
+}
+
+func (nMsgs StopAfter) configureMessages(opts *consumeOpts) error {
+	if nMsgs <= 0 {
+		return fmt.Errorf("%w: auto stop after value cannot be less than 1", ErrInvalidOption)
+	}
+	opts.StopAfter = int(nMsgs)
+	return nil
+}
+
 // ConsumeErrHandler sets custom error handler invoked when an error was encountered while consuming messages
 // It will be invoked for both terminal (Consumer Deleted, invalid request body) and non-terminal (e.g. missing heartbeats) errors
 func ConsumeErrHandler(cb ConsumeErrHandlerFunc) PullConsumeOpt {
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index 231b413..296ea6b 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -27,18 +27,21 @@ import (
 
 type (
 	orderedConsumer struct {
-		jetStream       *jetStream
-		cfg             *OrderedConsumerConfig
-		stream          string
-		currentConsumer *pullConsumer
-		cursor          cursor
-		namePrefix      string
-		serial          int
-		consumerType    consumerType
-		doReset         chan struct{}
-		resetInProgress uint32
-		userErrHandler  ConsumeErrHandlerFunc
-		runningFetch    *fetchResult
+		jetStream         *jetStream
+		cfg               *OrderedConsumerConfig
+		stream            string
+		currentConsumer   *pullConsumer
+		cursor            cursor
+		namePrefix        string
+		serial            int
+		consumerType      consumerType
+		doReset           chan struct{}
+		resetInProgress   uint32
+		userErrHandler    ConsumeErrHandlerFunc
+		stopAfter         int
+		stopAfterMsgsLeft chan int
+		withStopAfter     bool
+		runningFetch      *fetchResult
 		sync.Mutex
 	}
 
@@ -84,6 +87,18 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 	}
 	c.userErrHandler = consumeOpts.ErrHandler
 	opts = append(opts, ConsumeErrHandler(c.errHandler(c.serial)))
+	if consumeOpts.StopAfter > 0 {
+		c.withStopAfter = true
+		c.stopAfter = consumeOpts.StopAfter
+	}
+	c.stopAfterMsgsLeft = make(chan int, 1)
+	if c.stopAfter > 0 {
+		opts = append(opts, consumeStopAfterNotify(c.stopAfter, c.stopAfterMsgsLeft))
+	}
+	sub := &orderedSubscription{
+		consumer: c,
+		done:     make(chan struct{}, 1),
+	}
 	internalHandler := func(serial int) func(msg Msg) {
 		return func(msg Msg) {
 			// handler is a noop if message was delivered for a consumer with different serial
@@ -111,24 +126,45 @@ func (c *orderedConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt
 		return nil, err
 	}
 
-	sub := &orderedSubscription{
-		consumer: c,
-		done:     make(chan struct{}, 1),
-	}
 	go func() {
 		for {
 			select {
 			case <-c.doReset:
+				if c.withStopAfter {
+					select {
+					case c.stopAfter = <-c.stopAfterMsgsLeft:
+					default:
+					}
+					if c.stopAfter <= 0 {
+						sub.Stop()
+						return
+					}
+				}
 				if err := c.reset(); err != nil {
 					c.errHandler(c.serial)(c.currentConsumer.subscriptions[""], err)
 				}
+				if c.stopAfter > 0 {
+					opts = opts[:len(opts)-2]
+				} else {
+					opts = opts[:len(opts)-1]
+				}
+
 				// overwrite the previous err handler to use the new serial
-				opts[len(opts)-1] = ConsumeErrHandler(c.errHandler(c.serial))
+				opts = append(opts, ConsumeErrHandler(c.errHandler(c.serial)))
+				if c.withStopAfter {
+					opts = append(opts, consumeStopAfterNotify(c.stopAfter, c.stopAfterMsgsLeft))
+				}
 				if _, err := c.currentConsumer.Consume(internalHandler(c.serial), opts...); err != nil {
 					c.errHandler(c.serial)(c.currentConsumer.subscriptions[""], err)
 				}
 			case <-sub.done:
 				return
+			case msgsLeft, ok := <-c.stopAfterMsgsLeft:
+				if !ok {
+					close(sub.done)
+				}
+				c.stopAfter = msgsLeft
+				return
 			}
 		}
 	}()
@@ -143,12 +179,11 @@ func (c *orderedConsumer) errHandler(serial int) func(cc ConsumeContext, err err
 		if errors.Is(err, ErrNoHeartbeat) ||
 			errors.Is(err, errOrderedSequenceMismatch) ||
 			errors.Is(err, ErrConsumerDeleted) {
-			// only reset if serial matches the currect consumer serial and there is no reset in progress
+			// only reset if serial matches the current consumer serial and there is no reset in progress
 			if serial == c.serial && atomic.LoadUint32(&c.resetInProgress) == 0 {
 				atomic.StoreUint32(&c.resetInProgress, 1)
 				c.doReset <- struct{}{}
 			}
-
 		}
 	}
 }
@@ -171,8 +206,16 @@ func (c *orderedConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, er
 	if err != nil {
 		return nil, fmt.Errorf("%w: %s", ErrInvalidOption, err)
 	}
-	c.userErrHandler = consumeOpts.ErrHandler
 	opts = append(opts, WithMessagesErrOnMissingHeartbeat(true))
+	c.stopAfterMsgsLeft = make(chan int, 1)
+	if consumeOpts.StopAfter > 0 {
+		c.withStopAfter = true
+		c.stopAfter = consumeOpts.StopAfter
+	}
+	c.userErrHandler = consumeOpts.ErrHandler
+	if c.stopAfter > 0 {
+		opts = append(opts, messagesStopAfterNotify(c.stopAfter, c.stopAfterMsgsLeft))
+	}
 	_, err = c.currentConsumer.Messages(opts...)
 	if err != nil {
 		return nil, err
@@ -188,41 +231,55 @@ func (c *orderedConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, er
 }
 
 func (s *orderedSubscription) Next() (Msg, error) {
-	next := func() (Msg, error) {
-		for {
-			currentConsumer := s.consumer.currentConsumer
-			msg, err := currentConsumer.subscriptions[""].Next()
-			if err != nil {
-				if err := s.consumer.reset(); err != nil {
-					return nil, err
+	for {
+		currentConsumer := s.consumer.currentConsumer
+		msg, err := currentConsumer.subscriptions[""].Next()
+		if err != nil {
+			if errors.Is(err, ErrMsgIteratorClosed) {
+				s.Stop()
+				return nil, err
+			}
+			if s.consumer.withStopAfter {
+				select {
+				case s.consumer.stopAfter = <-s.consumer.stopAfterMsgsLeft:
+				default:
 				}
-				_, err := s.consumer.currentConsumer.Messages(s.opts...)
-				if err != nil {
-					return nil, err
+				if s.consumer.stopAfter <= 0 {
+					s.Stop()
+					return nil, ErrMsgIteratorClosed
 				}
-				continue
+				s.opts[len(s.opts)-1] = StopAfter(s.consumer.stopAfter)
 			}
-			meta, err := msg.Metadata()
-			if err != nil {
-				s.consumer.errHandler(s.consumer.serial)(currentConsumer.subscriptions[""], err)
-				continue
+			if err := s.consumer.reset(); err != nil {
+				return nil, err
 			}
-			serial := serialNumberFromConsumer(meta.Consumer)
-			dseq := meta.Sequence.Consumer
-			if dseq != s.consumer.cursor.deliverSeq+1 {
-				s.consumer.errHandler(serial)(currentConsumer.subscriptions[""], errOrderedSequenceMismatch)
-				continue
+			_, err := s.consumer.currentConsumer.Messages(s.opts...)
+			if err != nil {
+				return nil, err
 			}
-			s.consumer.cursor.deliverSeq = dseq
-			s.consumer.cursor.streamSeq = meta.Sequence.Stream
-			return msg, nil
+			continue
 		}
+		meta, err := msg.Metadata()
+		if err != nil {
+			s.consumer.errHandler(s.consumer.serial)(currentConsumer.subscriptions[""], err)
+			continue
+		}
+		serial := serialNumberFromConsumer(meta.Consumer)
+		dseq := meta.Sequence.Consumer
+		if dseq != s.consumer.cursor.deliverSeq+1 {
+			s.consumer.errHandler(serial)(currentConsumer.subscriptions[""], errOrderedSequenceMismatch)
+			continue
+		}
+		s.consumer.cursor.deliverSeq = dseq
+		s.consumer.cursor.streamSeq = meta.Sequence.Stream
+		return msg, nil
 	}
-	return next()
 }
 
 func (s *orderedSubscription) Stop() {
-	if s.consumer.currentConsumer == nil || s.consumer.currentConsumer.subscriptions[""] == nil {
+	s.consumer.currentConsumer.Lock()
+	defer s.consumer.currentConsumer.Unlock()
+	if s.consumer.currentConsumer.subscriptions[""] == nil {
 		return
 	}
 	s.consumer.currentConsumer.subscriptions[""].Stop()
@@ -329,7 +386,6 @@ func (c *orderedConsumer) reset() error {
 	defer c.Unlock()
 	defer atomic.StoreUint32(&c.resetInProgress, 0)
 	if c.currentConsumer != nil {
-		// c.currentConsumer.subscription.Stop()
 		var err error
 		for i := 0; ; i++ {
 			if c.cfg.MaxResetAttempts > 0 && i == c.cfg.MaxResetAttempts {
@@ -429,6 +485,22 @@ func (c *orderedConsumer) getConsumerConfigForSeq(seq uint64) *ConsumerConfig {
 	return cfg
 }
 
+func consumeStopAfterNotify(numMsgs int, msgsLeftAfterStop chan int) PullConsumeOpt {
+	return pullOptFunc(func(opts *consumeOpts) error {
+		opts.StopAfter = numMsgs
+		opts.stopAfterMsgsLeft = msgsLeftAfterStop
+		return nil
+	})
+}
+
+func messagesStopAfterNotify(numMsgs int, msgsLeftAfterStop chan int) PullMessagesOpt {
+	return pullOptFunc(func(opts *consumeOpts) error {
+		opts.StopAfter = numMsgs
+		opts.stopAfterMsgsLeft = msgsLeftAfterStop
+		return nil
+	})
+}
+
 func (c *orderedConsumer) Info(ctx context.Context) (*ConsumerInfo, error) {
 	c.Lock()
 	defer c.Unlock()
diff --git a/jetstream/pull.go b/jetstream/pull.go
index f3825d7..7029bb0 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -80,6 +80,8 @@ type (
 		ReportMissingHeartbeats bool
 		ThresholdMessages       int
 		ThresholdBytes          int
+		StopAfter               int
+		stopAfterMsgsLeft       chan int
 	}
 
 	ConsumeErrHandlerFunc func(consumeCtx ConsumeContext, err error)
@@ -99,6 +101,7 @@ type (
 		connStatusChanged chan nats.Status
 		fetchNext         chan *pullRequest
 		consumeOpts       *consumeOpts
+		delivered         int
 	}
 
 	pendingMsgs struct {
@@ -133,6 +136,13 @@ const (
 	unset              = -1
 )
 
+func min(x, y int) int {
+	if x < y {
+		return x
+	}
+	return y
+}
+
 // Consume returns a ConsumeContext, allowing for processing incoming messages from a stream in a given callback function.
 //
 // Available options:
@@ -214,7 +224,12 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 		handler(p.jetStream.toJSMsg(msg))
 		sub.Lock()
 		sub.decrementPendingMsgs(msg)
+		sub.incrementDeliveredMsgs()
 		sub.Unlock()
+
+		if sub.consumeOpts.StopAfter > 0 && sub.consumeOpts.StopAfter == sub.delivered {
+			sub.Stop()
+		}
 	}
 	inbox := p.jetStream.conn.NewInbox()
 	sub.subscription, err = p.jetStream.conn.Subscribe(inbox, internalHandler)
@@ -225,9 +240,13 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 	sub.Lock()
 	// initial pull
 	sub.resetPendingMsgs()
+	batchSize := sub.consumeOpts.MaxMessages
+	if sub.consumeOpts.StopAfter > 0 {
+		batchSize = min(batchSize, sub.consumeOpts.StopAfter-sub.delivered)
+	}
 	if err := sub.pull(&pullRequest{
 		Expires:   consumeOpts.Expires,
-		Batch:     consumeOpts.MaxMessages,
+		Batch:     batchSize,
 		MaxBytes:  consumeOpts.MaxBytes,
 		Heartbeat: consumeOpts.Heartbeat,
 	}, subject); err != nil {
@@ -276,10 +295,13 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 							}
 							time.Sleep(5 * time.Second)
 						}
-
+						batchSize := sub.consumeOpts.MaxMessages
+						if sub.consumeOpts.StopAfter > 0 {
+							batchSize = min(batchSize, sub.consumeOpts.StopAfter-sub.delivered)
+						}
 						sub.fetchNext <- &pullRequest{
 							Expires:   sub.consumeOpts.Expires,
-							Batch:     sub.consumeOpts.MaxMessages,
+							Batch:     batchSize,
 							MaxBytes:  sub.consumeOpts.MaxBytes,
 							Heartbeat: sub.consumeOpts.Heartbeat,
 						}
@@ -293,9 +315,13 @@ func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (
 					sub.consumeOpts.ErrHandler(sub, err)
 				}
 				if errors.Is(err, ErrNoHeartbeat) {
+					batchSize := sub.consumeOpts.MaxMessages
+					if sub.consumeOpts.StopAfter > 0 {
+						batchSize = min(batchSize, sub.consumeOpts.StopAfter-sub.delivered)
+					}
 					sub.fetchNext <- &pullRequest{
 						Expires:   sub.consumeOpts.Expires,
-						Batch:     sub.consumeOpts.MaxMessages,
+						Batch:     batchSize,
 						MaxBytes:  sub.consumeOpts.MaxBytes,
 						Heartbeat: sub.consumeOpts.Heartbeat,
 					}
@@ -328,6 +354,12 @@ func (s *pullSubscription) decrementPendingMsgs(msg *nats.Msg) {
 	}
 }
 
+// incrementDeliveredMsgs increments delivered message count
+// lock should be held before calling this method
+func (s *pullSubscription) incrementDeliveredMsgs() {
+	s.delivered++
+}
+
 // checkPending verifies whether there are enough messages in
 // the buffer to trigger a new pull request.
 // lock should be held before calling this method
@@ -335,16 +367,21 @@ func (s *pullSubscription) checkPending() {
 	if s.pending.msgCount < s.consumeOpts.ThresholdMessages ||
 		(s.pending.byteCount < s.consumeOpts.ThresholdBytes && s.consumeOpts.MaxBytes != 0) &&
 			atomic.LoadUint32(&s.fetchInProgress) == 1 {
-
-		s.fetchNext <- &pullRequest{
-			Expires:   s.consumeOpts.Expires,
-			Batch:     s.consumeOpts.MaxMessages - s.pending.msgCount,
-			MaxBytes:  s.consumeOpts.MaxBytes - s.pending.byteCount,
-			Heartbeat: s.consumeOpts.Heartbeat,
+		batchSize := s.consumeOpts.MaxMessages - s.pending.msgCount
+		if s.consumeOpts.StopAfter > 0 {
+			batchSize = min(batchSize, s.consumeOpts.StopAfter-s.delivered-s.pending.msgCount)
 		}
+		if batchSize > 0 {
+			s.fetchNext <- &pullRequest{
+				Expires:   s.consumeOpts.Expires,
+				Batch:     batchSize,
+				MaxBytes:  s.consumeOpts.MaxBytes - s.pending.byteCount,
+				Heartbeat: s.consumeOpts.Heartbeat,
+			}
 
-		s.pending.msgCount = s.consumeOpts.MaxMessages
-		s.pending.byteCount = s.consumeOpts.MaxBytes
+			s.pending.msgCount = s.consumeOpts.MaxMessages
+			s.pending.byteCount = s.consumeOpts.MaxBytes
+		}
 	}
 }
 
@@ -436,6 +473,11 @@ func (s *pullSubscription) Next() (Msg, error) {
 	}()
 
 	isConnected := true
+	if s.consumeOpts.StopAfter > 0 && s.delivered >= s.consumeOpts.StopAfter {
+		s.Stop()
+		return nil, ErrMsgIteratorClosed
+	}
+
 	for {
 		s.checkPending()
 		select {
@@ -458,6 +500,7 @@ func (s *pullSubscription) Next() (Msg, error) {
 				continue
 			}
 			s.decrementPendingMsgs(msg)
+			s.incrementDeliveredMsgs()
 			return s.consumer.jetStream.toJSMsg(msg), nil
 		case err := <-s.errs:
 			if errors.Is(err, ErrNoHeartbeat) {
@@ -551,6 +594,13 @@ func (s *pullSubscription) Stop() {
 		return
 	}
 	close(s.done)
+	if s.consumeOpts.stopAfterMsgsLeft != nil {
+		if s.delivered >= s.consumeOpts.StopAfter {
+			close(s.consumeOpts.stopAfterMsgsLeft)
+		} else {
+			s.consumeOpts.stopAfterMsgsLeft <- s.consumeOpts.StopAfter - s.delivered
+		}
+	}
 	atomic.StoreUint32(&s.closed, 1)
 }
 
@@ -572,7 +622,6 @@ func (p *pullConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, error)
 	}
 
 	return p.fetch(req)
-
 }
 
 // FetchBytes is used to retrieve up to a provided bytes from the stream.
@@ -780,6 +829,7 @@ func parseConsumeOpts(opts ...PullConsumeOpt) (*consumeOpts, error) {
 		Expires:                 DefaultExpires,
 		Heartbeat:               unset,
 		ReportMissingHeartbeats: true,
+		StopAfter:               unset,
 	}
 	for _, opt := range opts {
 		if err := opt.configureConsume(consumeOpts); err != nil {
@@ -799,6 +849,7 @@ func parseMessagesOpts(opts ...PullMessagesOpt) (*consumeOpts, error) {
 		Expires:                 DefaultExpires,
 		Heartbeat:               unset,
 		ReportMissingHeartbeats: true,
+		StopAfter:               unset,
 	}
 	for _, opt := range opts {
 		if err := opt.configureMessages(consumeOpts); err != nil {
diff --git a/js.go b/js.go
index a8deb98..ae19641 100644
--- a/js.go
+++ b/js.go
@@ -1176,10 +1176,11 @@ type SequencePair struct {
 
 // nextRequest is for getting next messages for pull based consumers from JetStream.
 type nextRequest struct {
-	Expires  time.Duration `json:"expires,omitempty"`
-	Batch    int           `json:"batch,omitempty"`
-	NoWait   bool          `json:"no_wait,omitempty"`
-	MaxBytes int           `json:"max_bytes,omitempty"`
+	Expires   time.Duration `json:"expires,omitempty"`
+	Batch     int           `json:"batch,omitempty"`
+	NoWait    bool          `json:"no_wait,omitempty"`
+	MaxBytes  int           `json:"max_bytes,omitempty"`
+	Heartbeat time.Duration `json:"idle_heartbeat,omitempty"`
 }
 
 // jsSub includes JetStream subscription info.
@@ -2469,6 +2470,7 @@ func EnableFlowControl() SubOpt {
 }
 
 // IdleHeartbeat enables push based consumers to have idle heartbeats delivered.
+// For pull consumers, idle heartbeat has to be set on each [Fetch] call.
 func IdleHeartbeat(duration time.Duration) SubOpt {
 	return subOptFn(func(opts *subOpts) error {
 		opts.cfg.Heartbeat = duration
@@ -2588,6 +2590,7 @@ type pullOpts struct {
 	maxBytes int
 	ttl      time.Duration
 	ctx      context.Context
+	hb       time.Duration
 }
 
 // PullOpt are the options that can be passed when pulling a batch of messages.
@@ -2603,6 +2606,16 @@ func PullMaxWaiting(n int) SubOpt {
 	})
 }
 
+type PullHeartbeat time.Duration
+
+func (h PullHeartbeat) configurePull(opts *pullOpts) error {
+	if h <= 0 {
+		return fmt.Errorf("%w: idle heartbeat has to be greater than 0", ErrInvalidArg)
+	}
+	opts.hb = time.Duration(h)
+	return nil
+}
+
 // PullMaxBytes defines the max bytes allowed for a fetch request.
 type PullMaxBytes int
 
@@ -2646,6 +2659,11 @@ func checkMsg(msg *Msg, checkSts, isNoWait bool) (usrMsg bool, err error) {
 	if !checkSts {
 		return
 	}
+
+	// if it's a heartbeat message, report as not user msg
+	if isHb, _ := isJSControlMessage(msg); isHb {
+		return
+	}
 	switch val {
 	case noResponders:
 		err = ErrNoResponders
@@ -2722,6 +2740,9 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		ttl = js.opts.wait
 	}
 	sub.mu.Unlock()
+	if o.hb != 0 && 2*o.hb >= ttl {
+		return nil, fmt.Errorf("%w: idle heartbeat value too large", ErrInvalidArg)
+	}
 
 	// Use the given context or setup a default one for the span
 	// of the pull batch request.
@@ -2732,7 +2753,6 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 	)
 	if ctx == nil {
 		ctx, cancel = context.WithTimeout(context.Background(), ttl)
-		defer cancel()
 	} else if _, hasDeadline := ctx.Deadline(); !hasDeadline {
 		// Prevent from passing the background context which will just block
 		// and cannot be canceled either.
@@ -2743,8 +2763,10 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		// If the context did not have a deadline, then create a new child context
 		// that will use the default timeout from the JS context.
 		ctx, cancel = context.WithTimeout(ctx, ttl)
-		defer cancel()
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
 	}
+	defer cancel()
 
 	// Check if context not done already before making the request.
 	select {
@@ -2783,6 +2805,8 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			msgs = append(msgs, msg)
 		}
 	}
+	var hbTimer *time.Timer
+	var hbErr error
 	if err == nil && len(msgs) < batch {
 		// For batch real size of 1, it does not make sense to set no_wait in
 		// the request.
@@ -2813,8 +2837,26 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			nr.Expires = expires
 			nr.NoWait = noWait
 			nr.MaxBytes = o.maxBytes
+			if 2*o.hb < expires {
+				nr.Heartbeat = o.hb
+			} else {
+				nr.Heartbeat = 0
+			}
 			req, _ := json.Marshal(nr)
-			return nc.PublishRequest(nms, rply, req)
+			if err := nc.PublishRequest(nms, rply, req); err != nil {
+				return err
+			}
+			if o.hb > 0 {
+				if hbTimer == nil {
+					hbTimer = time.AfterFunc(2*o.hb, func() {
+						hbErr = ErrNoHeartbeat
+						cancel()
+					})
+				} else {
+					hbTimer.Reset(2 * o.hb)
+				}
+			}
+			return nil
 		}
 
 		err = sendReq()
@@ -2822,6 +2864,9 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			// Ask for next message and wait if there are no messages
 			msg, err = sub.nextMsgWithContext(ctx, true, true)
 			if err == nil {
+				if hbTimer != nil {
+					hbTimer.Reset(2 * o.hb)
+				}
 				var usrMsg bool
 
 				usrMsg, err = checkMsg(msg, true, noWait)
@@ -2840,9 +2885,15 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 				}
 			}
 		}
+		if hbTimer != nil {
+			hbTimer.Stop()
+		}
 	}
 	// If there is at least a message added to msgs, then need to return OK and no error
 	if err != nil && len(msgs) == 0 {
+		if hbErr != nil {
+			return nil, hbErr
+		}
 		return nil, o.checkCtxErr(err)
 	}
 	return msgs, nil
@@ -2950,6 +3001,9 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 		ttl = js.opts.wait
 	}
 	sub.mu.Unlock()
+	if o.hb != 0 && 2*o.hb >= ttl {
+		return nil, fmt.Errorf("%w: idle heartbeat value too large", ErrInvalidArg)
+	}
 
 	// Use the given context or setup a default one for the span
 	// of the pull batch request.
@@ -2970,10 +3024,12 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 		// If the context did not have a deadline, then create a new child context
 		// that will use the default timeout from the JS context.
 		ctx, cancel = context.WithTimeout(ctx, ttl)
+	} else {
+		ctx, cancel = context.WithCancel(ctx)
 	}
 	defer func() {
 		// only cancel the context here if we are sure the fetching goroutine has not been started yet
-		if cancel != nil && cancelContext {
+		if cancelContext {
 			cancel()
 		}
 	}()
@@ -3031,9 +3087,10 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 
 	requestBatch := batch - len(result.msgs)
 	req := nextRequest{
-		Expires:  expires,
-		Batch:    requestBatch,
-		MaxBytes: o.maxBytes,
+		Expires:   expires,
+		Batch:     requestBatch,
+		MaxBytes:  o.maxBytes,
+		Heartbeat: o.hb,
 	}
 	reqJSON, err := json.Marshal(req)
 	if err != nil {
@@ -3051,11 +3108,17 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 		result.err = err
 		return result, nil
 	}
+	var hbTimer *time.Timer
+	var hbErr error
+	if o.hb > 0 {
+		hbTimer = time.AfterFunc(2*o.hb, func() {
+			hbErr = ErrNoHeartbeat
+			cancel()
+		})
+	}
 	cancelContext = false
 	go func() {
-		if cancel != nil {
-			defer cancel()
-		}
+		defer cancel()
 		var requestMsgs int
 		for requestMsgs < requestBatch {
 			// Ask for next message and wait if there are no messages
@@ -3063,6 +3126,9 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 			if err != nil {
 				break
 			}
+			if hbTimer != nil {
+				hbTimer.Reset(2 * o.hb)
+			}
 			var usrMsg bool
 
 			usrMsg, err = checkMsg(msg, true, false)
@@ -3082,7 +3148,11 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 			}
 		}
 		if err != nil {
-			result.err = o.checkCtxErr(err)
+			if hbErr != nil {
+				result.err = hbErr
+			} else {
+				result.err = o.checkCtxErr(err)
+			}
 		}
 		close(result.msgs)
 		result.done <- struct{}{}
diff --git a/jserrors.go b/jserrors.go
index d7959ca..cb3845c 100644
--- a/jserrors.go
+++ b/jserrors.go
@@ -104,6 +104,9 @@ var (
 	// ErrConsumerLeadershipChanged is returned when pending requests are no longer valid after leadership has changed
 	ErrConsumerLeadershipChanged JetStreamError = &jsError{message: "Leadership Changed"}
 
+	// ErrNoHeartbeat is returned when no heartbeat is received from server when sending requests with pull consumer.
+	ErrNoHeartbeat JetStreamError = &jsError{message: "no heartbeat received"}
+
 	// DEPRECATED: ErrInvalidDurableName is no longer returned and will be removed in future releases.
 	// Use ErrInvalidConsumerName instead.
 	ErrInvalidDurableName = errors.New("nats: invalid durable name")
diff --git a/micro/README.md b/micro/README.md
index c5af6f7..e4ffe09 100644
--- a/micro/README.md
+++ b/micro/README.md
@@ -95,6 +95,65 @@ _ = numbersGroup.AddEndpoint("add", micro.HandlerFunc(addHandler))
 _ = numbersGroup.AddEndpoint("multiply", micro.HandlerFunc(multiplyHandler))
 ```
 
+## Customizing queue groups
+
+For each service, group and endpoint the queue group used to gather responses
+can be customized. If not provided a default queue group will be used (`q`).
+Customizing queue groups can be useful to e.g. implement fanout request pattern
+or hedged request pattern (to reduce tail latencies by only waiting for the
+first response for multiple service instances).
+
+Let's say we have multiple services listening on the same subject, but with
+different queue groups:
+
+```go
+for i := 0; i < 5; i++ {
+  srv, _ := micro.AddService(nc, micro.Config{
+    Name:        "EchoService",
+    Version:     "1.0.0",
+    QueueGroup:  fmt.Sprintf("q-%d", i),
+    // base handler
+    Endpoint: &micro.EndpointConfig{
+        Subject: "svc.echo",
+        Handler: micro.HandlerFunc(echoHandler),
+    },
+  })
+}
+```
+
+In the client, we can send request to `svc.echo` to receive responses from all
+services registered on this subject (or wait only for the first response):
+
+```go
+sub, _ := nc.SubscribeSync("rply")
+nc.PublishRequest("svc.echo", "rply", nil)
+for start := time.Now(); time.Since(start) < 5*time.Second; {
+  msg, err := sub.NextMsg(1 * time.Second)
+  if err != nil {
+    break
+  }
+  fmt.Println("Received ", string(msg.Data))
+}
+```
+
+Queue groups can be overwritten by setting them on groups and endpoints as well:
+
+```go
+  srv, _ := micro.AddService(nc, micro.Config{
+    Name:        "EchoService",
+    Version:     "1.0.0",
+    QueueGroup:  "q1",
+  })
+
+  g := srv.AddGroup("g", micro.WithGroupQueueGroup("q2"))
+
+  // will be registered with queue group 'q2' from parent group
+  g.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}))
+
+  // will be registered with queue group 'q3'
+  g.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}), micro.WithEndpointQueueGroup("q3"))
+```
+
 ## Discovery and Monitoring
 
 Each service is assigned a unique ID on creation. A service instance is
diff --git a/micro/service.go b/micro/service.go
index 5d2dd36..a1067c8 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -39,7 +39,7 @@ type (
 
 		// AddGroup returns a Group interface, allowing for more complex endpoint topologies.
 		// A group can be used to register endpoints with given prefix.
-		AddGroup(string) Group
+		AddGroup(string, ...GroupOpt) Group
 
 		// Info returns the service info.
 		Info() Info
@@ -63,7 +63,7 @@ type (
 	// New groups can also be derived from a group using AddGroup.
 	Group interface {
 		// AddGroup creates a new group, prefixed by this group's prefix.
-		AddGroup(string) Group
+		AddGroup(string, ...GroupOpt) Group
 
 		// AddEndpoint registers new endpoints on a service.
 		// The endpoint's subject will be prefixed with the group prefix.
@@ -71,10 +71,16 @@ type (
 	}
 
 	EndpointOpt func(*endpointOpts) error
+	GroupOpt    func(*groupOpts) error
 
 	endpointOpts struct {
-		subject  string
-		metadata map[string]string
+		subject    string
+		metadata   map[string]string
+		queueGroup string
+	}
+
+	groupOpts struct {
+		queueGroup string
 	}
 
 	// ErrHandler is a function used to configure a custom error handler for a service,
@@ -108,6 +114,7 @@ type (
 	EndpointStats struct {
 		Name                  string          `json:"name"`
 		Subject               string          `json:"subject"`
+		QueueGroup            string          `json:"queue_group"`
 		NumRequests           int             `json:"num_requests"`
 		NumErrors             int             `json:"num_errors"`
 		LastError             string          `json:"last_error"`
@@ -131,9 +138,10 @@ type (
 	}
 
 	EndpointInfo struct {
-		Name     string            `json:"name"`
-		Subject  string            `json:"subject"`
-		Metadata map[string]string `json:"metadata"`
+		Name       string            `json:"name"`
+		Subject    string            `json:"subject"`
+		QueueGroup string            `json:"queue_group"`
+		Metadata   map[string]string `json:"metadata"`
 	}
 
 	// Endpoint manages a service endpoint.
@@ -148,8 +156,9 @@ type (
 	}
 
 	group struct {
-		service *service
-		prefix  string
+		service    *service
+		prefix     string
+		queueGroup string
 	}
 
 	// Verb represents a name of the monitoring service.
@@ -174,6 +183,9 @@ type (
 		// Metadata annotates the service
 		Metadata map[string]string `json:"metadata,omitempty"`
 
+		// QueueGroup can be used to override the default queue group name.
+		QueueGroup string `json:"queue_group"`
+
 		// StatsHandler is a user-defined custom function.
 		// used to calculate additional service stats.
 		StatsHandler StatsHandler
@@ -194,6 +206,9 @@ type (
 
 		// Metadata annotates the service
 		Metadata map[string]string `json:"metadata,omitempty"`
+
+		// QueueGroup can be used to override the default queue group name.
+		QueueGroup string `json:"queue_group"`
 	}
 
 	// NATSError represents an error returned by a NATS Subscription.
@@ -235,7 +250,7 @@ type (
 
 const (
 	// Queue Group name used across all services
-	QG = "q"
+	DefaultQueueGroup = "q"
 
 	// APIPrefix is the root of all control subjects
 	APIPrefix = "$SRV"
@@ -330,6 +345,11 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 		if config.Endpoint.Metadata != nil {
 			opts = append(opts, WithEndpointMetadata(config.Endpoint.Metadata))
 		}
+		if config.Endpoint.QueueGroup != "" {
+			opts = append(opts, WithEndpointQueueGroup(config.Endpoint.QueueGroup))
+		} else if config.QueueGroup != "" {
+			opts = append(opts, WithEndpointQueueGroup(config.QueueGroup))
+		}
 		if err := svc.AddEndpoint("default", config.Endpoint.Handler, opts...); err != nil {
 			svc.asyncDispatcher.close()
 			return nil, err
@@ -380,29 +400,34 @@ func (s *service) AddEndpoint(name string, handler Handler, opts ...EndpointOpt)
 	if options.subject != "" {
 		subject = options.subject
 	}
-
-	return addEndpoint(s, name, subject, handler, options.metadata)
+	queueGroup := queueGroupName(options.queueGroup, s.Config.QueueGroup)
+	return addEndpoint(s, name, subject, handler, options.metadata, queueGroup)
 }
 
-func addEndpoint(s *service, name, subject string, handler Handler, metadata map[string]string) error {
+func addEndpoint(s *service, name, subject string, handler Handler, metadata map[string]string, queueGroup string) error {
 	if !nameRegexp.MatchString(name) {
 		return fmt.Errorf("%w: invalid endpoint name", ErrConfigValidation)
 	}
 	if !subjectRegexp.MatchString(subject) {
 		return fmt.Errorf("%w: invalid endpoint subject", ErrConfigValidation)
 	}
+	if !subjectRegexp.MatchString(queueGroup) {
+		return fmt.Errorf("%w: invalid endpoint queue group", ErrConfigValidation)
+	}
 	endpoint := &Endpoint{
 		service: s,
 		EndpointConfig: EndpointConfig{
-			Subject:  subject,
-			Handler:  handler,
-			Metadata: metadata,
+			Subject:    subject,
+			Handler:    handler,
+			Metadata:   metadata,
+			QueueGroup: queueGroup,
 		},
 		Name: name,
 	}
+
 	sub, err := s.nc.QueueSubscribe(
 		subject,
-		QG,
+		queueGroup,
 		func(m *nats.Msg) {
 			s.reqHandler(endpoint, &request{msg: m})
 		},
@@ -413,16 +438,23 @@ func addEndpoint(s *service, name, subject string, handler Handler, metadata map
 	endpoint.subscription = sub
 	s.endpoints = append(s.endpoints, endpoint)
 	endpoint.stats = EndpointStats{
-		Name:    name,
-		Subject: subject,
+		Name:       name,
+		Subject:    subject,
+		QueueGroup: queueGroup,
 	}
 	return nil
 }
 
-func (s *service) AddGroup(name string) Group {
+func (s *service) AddGroup(name string, opts ...GroupOpt) Group {
+	var o groupOpts
+	for _, opt := range opts {
+		opt(&o)
+	}
+	queueGroup := queueGroupName(o.queueGroup, s.Config.QueueGroup)
 	return &group{
-		service: s,
-		prefix:  name,
+		service:    s,
+		prefix:     name,
+		queueGroup: queueGroup,
 	}
 }
 
@@ -448,11 +480,15 @@ func (ac *asyncCallbacksHandler) close() {
 
 func (c *Config) valid() error {
 	if !nameRegexp.MatchString(c.Name) {
-		return fmt.Errorf("%w: service name: name should not be empty and should consist of alphanumerical charactest, dashes and underscores", ErrConfigValidation)
+		return fmt.Errorf("%w: service name: name should not be empty and should consist of alphanumerical characters, dashes and underscores", ErrConfigValidation)
 	}
 	if !semVerRegexp.MatchString(c.Version) {
 		return fmt.Errorf("%w: version: version should not be empty should match the SemVer format", ErrConfigValidation)
 	}
+	if c.QueueGroup != "" && !subjectRegexp.MatchString(c.QueueGroup) {
+		return fmt.Errorf("%w: queue group: invalid queue group name", ErrConfigValidation)
+	}
+
 	return nil
 }
 
@@ -658,9 +694,10 @@ func (s *service) Info() Info {
 	endpoints := make([]EndpointInfo, 0, len(s.endpoints))
 	for _, e := range s.endpoints {
 		endpoints = append(endpoints, EndpointInfo{
-			Name:     e.Name,
-			Subject:  e.Subject,
-			Metadata: e.Metadata,
+			Name:       e.Name,
+			Subject:    e.Subject,
+			QueueGroup: e.QueueGroup,
+			Metadata:   e.Metadata,
 		})
 	}
 
@@ -687,6 +724,7 @@ func (s *service) Stats() Stats {
 		endpointStats := &EndpointStats{
 			Name:                  endpoint.stats.Name,
 			Subject:               endpoint.stats.Subject,
+			QueueGroup:            endpoint.stats.QueueGroup,
 			NumRequests:           endpoint.stats.NumRequests,
 			NumErrors:             endpoint.stats.NumErrors,
 			LastError:             endpoint.stats.LastError,
@@ -738,10 +776,30 @@ func (g *group) AddEndpoint(name string, handler Handler, opts ...EndpointOpt) e
 	if g.prefix == "" {
 		endpointSubject = subject
 	}
-	return addEndpoint(g.service, name, endpointSubject, handler, options.metadata)
+	queueGroup := queueGroupName(options.queueGroup, g.queueGroup)
+
+	return addEndpoint(g.service, name, endpointSubject, handler, options.metadata, queueGroup)
 }
 
-func (g *group) AddGroup(name string) Group {
+func queueGroupName(customQG, parentQG string) string {
+	queueGroup := customQG
+	if queueGroup == "" {
+		if parentQG != "" {
+			queueGroup = parentQG
+		} else {
+			queueGroup = DefaultQueueGroup
+		}
+	}
+	return queueGroup
+}
+
+func (g *group) AddGroup(name string, opts ...GroupOpt) Group {
+	var o groupOpts
+	for _, opt := range opts {
+		opt(&o)
+	}
+	queueGroup := queueGroupName(o.queueGroup, g.queueGroup)
+
 	parts := make([]string, 0, 2)
 	if g.prefix != "" {
 		parts = append(parts, g.prefix)
@@ -752,8 +810,9 @@ func (g *group) AddGroup(name string) Group {
 	prefix := strings.Join(parts, ".")
 
 	return &group{
-		service: g.service,
-		prefix:  prefix,
+		service:    g.service,
+		prefix:     prefix,
+		queueGroup: queueGroup,
 	}
 }
 
@@ -817,3 +876,17 @@ func WithEndpointMetadata(metadata map[string]string) EndpointOpt {
 		return nil
 	}
 }
+
+func WithEndpointQueueGroup(queueGroup string) EndpointOpt {
+	return func(e *endpointOpts) error {
+		e.queueGroup = queueGroup
+		return nil
+	}
+}
+
+func WithGroupQueueGroup(queueGroup string) GroupOpt {
+	return func(g *groupOpts) error {
+		g.queueGroup = queueGroup
+		return nil
+	}
+}
diff --git a/nats.go b/nats.go
index 36fb9df..821e823 100644
--- a/nats.go
+++ b/nats.go
@@ -3168,8 +3168,10 @@ func (nc *Conn) processMsg(data []byte) {
 		}
 	}
 
-	// Skip processing if this is a control message.
-	if !ctrlMsg {
+	// Skip processing if this is a control message and
+	// if not a pull consumer heartbeat. For pull consumers,
+	// heartbeats have to be handled on per request basis.
+	if !ctrlMsg || (jsi != nil && jsi.pull) {
 		var chanSubCheckFC bool
 		// Subscription internal stats (applicable only for non ChanSubscription's)
 		if sub.typ != ChanSubscription {


Test patch:
diff --git a/jetstream/test/ordered_test.go b/jetstream/test/ordered_test.go
index bb35843..948caf2 100644
--- a/jetstream/test/ordered_test.go
+++ b/jetstream/test/ordered_test.go
@@ -155,6 +155,136 @@ func TestOrderedConsumerConsume(t *testing.T) {
 			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrOrderedConsumerConcurrentRequests, err)
 		}
 	})
+
+	t.Run("with auto unsubscribe", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		msgs := make([]jetstream.Msg, 0)
+		wg := &sync.WaitGroup{}
+		wg.Add(50)
+		_, err = c.Consume(func(msg jetstream.Msg) {
+			msgs = append(msgs, msg)
+			msg.Ack()
+			wg.Done()
+		}, jetstream.StopAfter(50), jetstream.PullMaxMessages(40))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		wg.Wait()
+		time.Sleep(10 * time.Millisecond)
+		ci, err := c.Info(ctx)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ci.NumPending != 50 {
+			t.Fatalf("Unexpected number of pending messages; want 50; got %d", ci.NumPending)
+		}
+		if ci.NumAckPending != 0 {
+			t.Fatalf("Unexpected number of ack pending messages; want 0; got %d", ci.NumAckPending)
+		}
+		if ci.NumWaiting != 0 {
+			t.Fatalf("Unexpected number of waiting pull requests; want 0; got %d", ci.NumWaiting)
+		}
+		time.Sleep(10 * time.Millisecond)
+	})
+
+	t.Run("with auto unsubscribe and consumer reset", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		msgs := make([]jetstream.Msg, 0)
+		wg := &sync.WaitGroup{}
+		wg.Add(100)
+		_, err = c.Consume(func(msg jetstream.Msg) {
+			msgs = append(msgs, msg)
+			msg.Ack()
+			wg.Done()
+		}, jetstream.StopAfter(150), jetstream.PullMaxMessages(40))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		wg.Wait()
+		if err := s.DeleteConsumer(ctx, c.CachedInfo().Name); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		wg.Add(50)
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		wg.Wait()
+
+		time.Sleep(10 * time.Millisecond)
+		ci, err := c.Info(ctx)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ci.NumPending != 50 {
+			t.Fatalf("Unexpected number of pending messages; want 50; got %d", ci.NumPending)
+		}
+		if ci.NumAckPending != 0 {
+			t.Fatalf("Unexpected number of ack pending messages; want 0; got %d", ci.NumAckPending)
+		}
+		if ci.NumWaiting != 0 {
+			t.Fatalf("Unexpected number of waiting pull requests; want 0; got %d", ci.NumWaiting)
+		}
+		time.Sleep(10 * time.Millisecond)
+	})
 }
 
 func TestOrderedConsumerMessages(t *testing.T) {
@@ -224,6 +354,168 @@ func TestOrderedConsumerMessages(t *testing.T) {
 		}
 	})
 
+	t.Run("with auto unsubscribe", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "test", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		msgs := make([]jetstream.Msg, 0)
+		it, err := c.Messages(jetstream.StopAfter(50), jetstream.PullMaxMessages(40))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		for i := 0; i < 50; i++ {
+			msg, err := it.Next()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if msg == nil {
+				break
+			}
+			if err := msg.Ack(); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			msgs = append(msgs, msg)
+
+		}
+		if _, err := it.Next(); err != jetstream.ErrMsgIteratorClosed {
+			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrMsgIteratorClosed, err)
+		}
+		if len(msgs) != 50 {
+			t.Fatalf("Unexpected received message count; want %d; got %d", 50, len(msgs))
+		}
+		ci, err := c.Info(ctx)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ci.NumPending != 50 {
+			t.Fatalf("Unexpected number of pending messages; want 50; got %d", ci.NumPending)
+		}
+		if ci.NumAckPending != 0 {
+			t.Fatalf("Unexpected number of ack pending messages; want 0; got %d", ci.NumAckPending)
+		}
+		if ci.NumWaiting != 0 {
+			t.Fatalf("Unexpected number of waiting pull requests; want 0; got %d", ci.NumWaiting)
+		}
+	})
+
+	t.Run("with auto unsubscribe and consumer reset", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "test", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		msgs := make([]jetstream.Msg, 0)
+		it, err := c.Messages(jetstream.StopAfter(150), jetstream.PullMaxMessages(40))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		for i := 0; i < 100; i++ {
+			msg, err := it.Next()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if msg == nil {
+				break
+			}
+			if err := msg.Ack(); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			msgs = append(msgs, msg)
+		}
+		if err := s.DeleteConsumer(ctx, c.CachedInfo().Name); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		for i := 0; i < 50; i++ {
+			msg, err := it.Next()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if msg == nil {
+				break
+			}
+			if err := msg.Ack(); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			msgs = append(msgs, msg)
+		}
+		if _, err := it.Next(); err != jetstream.ErrMsgIteratorClosed {
+			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrMsgIteratorClosed, err)
+		}
+		if len(msgs) != 150 {
+			t.Fatalf("Unexpected received message count; want %d; got %d", 50, len(msgs))
+		}
+		ci, err := c.Info(ctx)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ci.NumPending != 50 {
+			t.Fatalf("Unexpected number of pending messages; want 50; got %d", ci.NumPending)
+		}
+		if ci.NumAckPending != 0 {
+			t.Fatalf("Unexpected number of ack pending messages; want 0; got %d", ci.NumAckPending)
+		}
+		if ci.NumWaiting != 0 {
+			t.Fatalf("Unexpected number of waiting pull requests; want 0; got %d", ci.NumWaiting)
+		}
+	})
+
 	t.Run("consumer used as fetch", func(t *testing.T) {
 		srv := RunBasicJetStreamServer()
 		defer shutdownJSServerAndRemoveStorage(t, srv)
diff --git a/jetstream/test/pull_test.go b/jetstream/test/pull_test.go
index 803784b..981a44f 100644
--- a/jetstream/test/pull_test.go
+++ b/jetstream/test/pull_test.go
@@ -1051,6 +1051,77 @@ func TestPullConsumerMessages(t *testing.T) {
 		}
 	})
 
+	t.Run("with auto unsubscribe", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "test", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		msgs := make([]jetstream.Msg, 0)
+		it, err := c.Messages(jetstream.StopAfter(50), jetstream.PullMaxMessages(40))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		for i := 0; i < 50; i++ {
+			msg, err := it.Next()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if msg == nil {
+				break
+			}
+			if err := msg.DoubleAck(ctx); err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			msgs = append(msgs, msg)
+
+		}
+		if _, err := it.Next(); err != jetstream.ErrMsgIteratorClosed {
+			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrMsgIteratorClosed, err)
+		}
+		if len(msgs) != 50 {
+			t.Fatalf("Unexpected received message count; want %d; got %d", 50, len(msgs))
+		}
+		ci, err := c.Info(ctx)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ci.NumPending != 50 {
+			t.Fatalf("Unexpected number of pending messages; want 50; got %d", ci.NumPending)
+		}
+		if ci.NumAckPending != 0 {
+			t.Fatalf("Unexpected number of ack pending messages; want 0; got %d", ci.NumAckPending)
+		}
+		if ci.NumWaiting != 0 {
+			t.Fatalf("Unexpected number of waiting pull requests; want 0; got %d", ci.NumWaiting)
+		}
+	})
+
 	t.Run("create iterator, stop, then create again", func(t *testing.T) {
 		srv := RunBasicJetStreamServer()
 		defer shutdownJSServerAndRemoveStorage(t, srv)
@@ -1703,6 +1774,65 @@ func TestPullConsumerConsume(t *testing.T) {
 		}
 	})
 
+	t.Run("with auto unsubscribe", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish(ctx, "FOO.A", []byte("msg")); err != nil {
+				t.Fatalf("Unexpected error during publish: %s", err)
+			}
+		}
+		msgs := make([]jetstream.Msg, 0)
+		wg := &sync.WaitGroup{}
+		wg.Add(50)
+		_, err = c.Consume(func(msg jetstream.Msg) {
+			msgs = append(msgs, msg)
+			msg.Ack()
+			wg.Done()
+		}, jetstream.StopAfter(50), jetstream.PullMaxMessages(40))
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		wg.Wait()
+		time.Sleep(10 * time.Millisecond)
+		ci, err := c.Info(ctx)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if ci.NumPending != 50 {
+			t.Fatalf("Unexpected number of pending messages; want 50; got %d", ci.NumPending)
+		}
+		if ci.NumAckPending != 0 {
+			t.Fatalf("Unexpected number of ack pending messages; want 0; got %d", ci.NumAckPending)
+		}
+		if ci.NumWaiting != 0 {
+			t.Fatalf("Unexpected number of waiting pull requests; want 0; got %d", ci.NumWaiting)
+		}
+	})
+
 	t.Run("with invalid batch size", func(t *testing.T) {
 		srv := RunBasicJetStreamServer()
 		defer shutdownJSServerAndRemoveStorage(t, srv)
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index 5386ae2..d697c59 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -664,8 +664,9 @@ func TestGroups(t *testing.T) {
 			name:         "no groups",
 			endpointName: "foo",
 			expectedEndpoint: micro.EndpointInfo{
-				Name:    "foo",
-				Subject: "foo",
+				Name:       "foo",
+				Subject:    "foo",
+				QueueGroup: "q",
 			},
 		},
 		{
@@ -673,8 +674,9 @@ func TestGroups(t *testing.T) {
 			endpointName: "foo",
 			groups:       []string{"g1"},
 			expectedEndpoint: micro.EndpointInfo{
-				Name:    "foo",
-				Subject: "g1.foo",
+				Name:       "foo",
+				Subject:    "g1.foo",
+				QueueGroup: "q",
 			},
 		},
 		{
@@ -682,8 +684,9 @@ func TestGroups(t *testing.T) {
 			endpointName: "foo",
 			groups:       []string{""},
 			expectedEndpoint: micro.EndpointInfo{
-				Name:    "foo",
-				Subject: "foo",
+				Name:       "foo",
+				Subject:    "foo",
+				QueueGroup: "q",
 			},
 		},
 		{
@@ -691,8 +694,9 @@ func TestGroups(t *testing.T) {
 			endpointName: "foo",
 			groups:       []string{"", "g1", ""},
 			expectedEndpoint: micro.EndpointInfo{
-				Name:    "foo",
-				Subject: "g1.foo",
+				Name:       "foo",
+				Subject:    "g1.foo",
+				QueueGroup: "q",
 			},
 		},
 		{
@@ -700,8 +704,9 @@ func TestGroups(t *testing.T) {
 			endpointName: "foo",
 			groups:       []string{"g1", "g2", "g3"},
 			expectedEndpoint: micro.EndpointInfo{
-				Name:    "foo",
-				Subject: "g1.g2.g3.foo",
+				Name:       "foo",
+				Subject:    "g1.g2.g3.foo",
+				QueueGroup: "q",
 			},
 		},
 	}
@@ -849,9 +854,10 @@ func TestMonitoringHandlers(t *testing.T) {
 				},
 				Endpoints: []micro.EndpointInfo{
 					{
-						Name:     "default",
-						Subject:  "test.func",
-						Metadata: map[string]string{"basic": "schema"},
+						Name:       "default",
+						Subject:    "test.func",
+						QueueGroup: "q",
+						Metadata:   map[string]string{"basic": "schema"},
 					},
 				},
 			},
@@ -869,9 +875,10 @@ func TestMonitoringHandlers(t *testing.T) {
 				},
 				Endpoints: []micro.EndpointInfo{
 					{
-						Name:     "default",
-						Subject:  "test.func",
-						Metadata: map[string]string{"basic": "schema"},
+						Name:       "default",
+						Subject:    "test.func",
+						QueueGroup: "q",
+						Metadata:   map[string]string{"basic": "schema"},
 					},
 				},
 			},
@@ -889,9 +896,10 @@ func TestMonitoringHandlers(t *testing.T) {
 				},
 				Endpoints: []micro.EndpointInfo{
 					{
-						Name:     "default",
-						Subject:  "test.func",
-						Metadata: map[string]string{"basic": "schema"},
+						Name:       "default",
+						Subject:    "test.func",
+						QueueGroup: "q",
+						Metadata:   map[string]string{"basic": "schema"},
 					},
 				},
 			},
@@ -1431,3 +1439,305 @@ func TestControlSubject(t *testing.T) {
 		})
 	}
 }
+
+func TestCustomQueueGroup(t *testing.T) {
+	tests := []struct {
+		name                string
+		endpointInit        func(*testing.T, *nats.Conn) micro.Service
+		expectedQueueGroups map[string]string
+	}{
+		{
+			name: "default queue group",
+			endpointInit: func(t *testing.T, nc *nats.Conn) micro.Service {
+				srv, err := micro.AddService(nc, micro.Config{
+					Name:    "test_service",
+					Version: "0.0.1",
+					Endpoint: &micro.EndpointConfig{
+						Subject: "foo",
+						Handler: micro.HandlerFunc(func(r micro.Request) {}),
+					},
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				err = srv.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				return srv
+			},
+			expectedQueueGroups: map[string]string{
+				"default": "q",
+				"bar":     "q",
+			},
+		},
+		{
+			name: "custom queue group on service config",
+			endpointInit: func(t *testing.T, nc *nats.Conn) micro.Service {
+				srv, err := micro.AddService(nc, micro.Config{
+					Name:       "test_service",
+					Version:    "0.0.1",
+					QueueGroup: "custom",
+					Endpoint: &micro.EndpointConfig{
+						Subject: "foo",
+						Handler: micro.HandlerFunc(func(r micro.Request) {}),
+					},
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				// add endpoint on service directly, should have the same queue group
+				err = srv.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				// add group with queue group from service config
+				g1 := srv.AddGroup("g1")
+
+				// add endpoint on group, should have queue group from service config
+				err = g1.AddEndpoint("baz", micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				return srv
+			},
+			expectedQueueGroups: map[string]string{
+				"default": "custom",
+				"bar":     "custom",
+				"baz":     "custom",
+			},
+		},
+		{
+			name: "overwriting queue groups",
+			endpointInit: func(t *testing.T, nc *nats.Conn) micro.Service {
+				srv, err := micro.AddService(nc, micro.Config{
+					Name:       "test_service",
+					Version:    "0.0.1",
+					QueueGroup: "q-config",
+					Endpoint: &micro.EndpointConfig{
+						Subject:    "foo",
+						QueueGroup: "q-default",
+						Handler:    micro.HandlerFunc(func(r micro.Request) {}),
+					},
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				g1 := srv.AddGroup("g1", micro.WithGroupQueueGroup("q-g1"))
+
+				// should have the same queue group as the parent group
+				g2 := g1.AddGroup("g2")
+
+				// overwrite parent group queue group
+				g3 := g2.AddGroup("g3", micro.WithGroupQueueGroup("q-g3"))
+
+				// add endpoint on service directly, overwriting the queue group
+				err = srv.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}), micro.WithEndpointQueueGroup("q-bar"))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				// add endpoint on group, should have queue group from g1
+				err = g2.AddEndpoint("baz", micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				// add endpoint on group, overwriting the queue group
+				err = g2.AddEndpoint("qux", micro.HandlerFunc(func(r micro.Request) {}), micro.WithEndpointQueueGroup("q-qux"))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				// add endpoint on group, should have queue group from g3
+				err = g3.AddEndpoint("quux", micro.HandlerFunc(func(r micro.Request) {}))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				return srv
+			},
+			expectedQueueGroups: map[string]string{
+				"default": "q-default",
+				"bar":     "q-bar",
+				"baz":     "q-g1",
+				"qux":     "q-qux",
+				"quux":    "q-g3",
+			},
+		},
+		{
+			name: "empty queue group in option, inherit from parent",
+			endpointInit: func(t *testing.T, nc *nats.Conn) micro.Service {
+				srv, err := micro.AddService(nc, micro.Config{
+					Name:       "test_service",
+					Version:    "0.0.1",
+					QueueGroup: "q-config",
+				})
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				// add endpoint on service directly, overwriting the queue group
+				err = srv.AddEndpoint("bar", micro.HandlerFunc(func(r micro.Request) {}), micro.WithEndpointQueueGroup(""))
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+
+				return srv
+			},
+			expectedQueueGroups: map[string]string{
+				"bar": "q-config",
+			},
+		},
+		{
+			name: "invalid queue group on service config",
+			endpointInit: func(t *testing.T, nc *nats.Conn) micro.Service {
+				_, err := micro.AddService(nc, micro.Config{
+					Name:       "test_service",
+					Version:    "0.0.1",
+					QueueGroup: ">.abc",
+					Endpoint: &micro.EndpointConfig{
+						Subject: "foo",
+						Handler: micro.HandlerFunc(func(r micro.Request) {}),
+					},
+				})
+				if !errors.Is(err, micro.ErrConfigValidation) {
+					t.Fatalf("Expected error: %v; got: %v", micro.ErrConfigValidation, err)
+				}
+				return nil
+			},
+		},
+		{
+			name: "invalid queue group on endpoint",
+			endpointInit: func(t *testing.T, nc *nats.Conn) micro.Service {
+				_, err := micro.AddService(nc, micro.Config{
+					Name:    "test_service",
+					Version: "0.0.1",
+					Endpoint: &micro.EndpointConfig{
+						Subject:    "foo",
+						QueueGroup: ">.abc",
+						Handler:    micro.HandlerFunc(func(r micro.Request) {}),
+					},
+				})
+				if !errors.Is(err, micro.ErrConfigValidation) {
+					t.Fatalf("Expected error: %v; got: %v", micro.ErrConfigValidation, err)
+				}
+				return nil
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := RunServerOnPort(-1)
+			defer s.Shutdown()
+
+			nc, err := nats.Connect(s.ClientURL())
+			if err != nil {
+				t.Fatalf("Expected to connect to server, got %v", err)
+			}
+			defer nc.Close()
+			srv := test.endpointInit(t, nc)
+			if srv == nil {
+				return
+			}
+			defer srv.Stop()
+			info := srv.Info()
+			endpoints := make(map[string]micro.EndpointInfo)
+			for _, e := range info.Endpoints {
+				endpoints[e.Name] = e
+			}
+			if len(endpoints) != len(test.expectedQueueGroups) {
+				t.Fatalf("Expected %d endpoints; got: %d", len(test.expectedQueueGroups), len(endpoints))
+			}
+			for name, expectedGroup := range test.expectedQueueGroups {
+				if endpoints[name].QueueGroup != expectedGroup {
+					t.Fatalf("Invalid queue group for endpoint %q; want: %q; got: %q", name, expectedGroup, endpoints[name].QueueGroup)
+				}
+			}
+
+			stats := srv.Stats()
+			// make sure the same queue groups are on stats
+			endpointStats := make(map[string]*micro.EndpointStats)
+
+			for _, e := range stats.Endpoints {
+				endpointStats[e.Name] = e
+			}
+			if len(endpointStats) != len(test.expectedQueueGroups) {
+				t.Fatalf("Expected %d endpoints; got: %d", len(test.expectedQueueGroups), len(endpointStats))
+			}
+			for name, expectedGroup := range test.expectedQueueGroups {
+				if endpointStats[name].QueueGroup != expectedGroup {
+					t.Fatalf("Invalid queue group for endpoint %q; want: %q; got: %q", name, expectedGroup, endpointStats[name].QueueGroup)
+				}
+			}
+		})
+	}
+}
+
+func TestCustomQueueGroupMultipleResponses(t *testing.T) {
+	s := RunServerOnPort(-1)
+	defer s.Shutdown()
+
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatalf("Expected to connect to server, got %v", err)
+	}
+	defer nc.Close()
+
+	for i := 0; i < 5; i++ {
+		f := func(i int) func(r micro.Request) {
+			return func(r micro.Request) {
+				time.Sleep(10 * time.Millisecond)
+				r.Respond([]byte(fmt.Sprintf("%d", i)))
+			}
+		}
+		service, err := micro.AddService(nc, micro.Config{
+			Name:       "test_service",
+			Version:    "0.0.1",
+			QueueGroup: fmt.Sprintf("q-%d", i),
+			Endpoint: &micro.EndpointConfig{
+				Subject: "foo",
+				Handler: micro.HandlerFunc(f(i)),
+			},
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer service.Stop()
+	}
+	err = nc.PublishRequest("foo", "rply", []byte("req"))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	sub, err := nc.SubscribeSync("rply")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	expectedResponses := map[string]bool{
+		"0": false,
+		"1": false,
+		"2": false,
+		"3": false,
+		"4": false,
+	}
+	defer sub.Unsubscribe()
+	for i := 0; i < 5; i++ {
+		msg, err := sub.NextMsg(1 * time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		expectedResponses[string(msg.Data)] = true
+	}
+	msg, err := sub.NextMsg(100 * time.Millisecond)
+	if err == nil {
+		t.Fatalf("Unexpected message: %v", string(msg.Data))
+	}
+	for k, v := range expectedResponses {
+		if !v {
+			t.Fatalf("Did not receive response from service %s", k)
+		}
+	}
+}
diff --git a/test/js_test.go b/test/js_test.go
index d9ad8ca..ab599c4 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1124,6 +1124,176 @@ func TestJetStreamSubscribe_SkipConsumerLookup(t *testing.T) {
 	})
 }
 
+func TestPullSubscribeFetchWithHeartbeat(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sub, err := js.PullSubscribe("foo", "")
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	for i := 0; i < 5; i++ {
+		if _, err := js.Publish("foo", []byte("msg")); err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+	}
+
+	// fetch 5 messages, should finish immediately
+	msgs, err := sub.Fetch(5, nats.PullHeartbeat(100*time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	if len(msgs) != 5 {
+		t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+	}
+	now := time.Now()
+	// no messages available, should time out normally
+	_, err = sub.Fetch(5, nats.PullHeartbeat(50*time.Millisecond), nats.MaxWait(300*time.Millisecond))
+	elapsed := time.Since(now)
+	if elapsed < 300*time.Millisecond {
+		t.Fatalf("Expected timeout after 300ms; got: %v", elapsed)
+	}
+	if !errors.Is(err, nats.ErrTimeout) {
+		t.Fatalf("Expected timeout error; got: %v", err)
+	}
+
+	// delete consumer to verify heartbeats are not sent anymore
+	info, err := sub.ConsumerInfo()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if err := js.DeleteConsumer("TEST", info.Name); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	_, err = sub.Fetch(5, nats.PullHeartbeat(100*time.Millisecond), nats.MaxWait(1*time.Second))
+	if !errors.Is(err, nats.ErrNoHeartbeat) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+
+	// heartbeat value too large
+	_, err = sub.Fetch(5, nats.PullHeartbeat(200*time.Millisecond), nats.MaxWait(300*time.Millisecond))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+
+	// heartbeat value invalid
+	_, err = sub.Fetch(5, nats.PullHeartbeat(-1))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+}
+
+func TestPullSubscribeFetchBatchWithHeartbeat(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sub, err := js.PullSubscribe("foo", "")
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	for i := 0; i < 5; i++ {
+		if _, err := js.Publish("foo", []byte("msg")); err != nil {
+			t.Fatalf("Unexpected error: %s", err)
+		}
+	}
+
+	// fetch 5 messages, should finish immediately
+	msgs, err := sub.FetchBatch(5, nats.PullHeartbeat(100*time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	var i int
+	for msg := range msgs.Messages() {
+		i++
+		msg.Ack()
+	}
+	if i != 5 {
+		t.Fatalf("Expected %d messages; got: %d", 5, i)
+	}
+	if msgs.Error() != nil {
+		t.Fatalf("Unexpected error: %s", msgs.Error())
+	}
+	now := time.Now()
+	// no messages available, should time out normally
+	msgs, err = sub.FetchBatch(5, nats.PullHeartbeat(50*time.Millisecond), nats.MaxWait(300*time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	i = 0
+	for msg := range msgs.Messages() {
+		i++
+		msg.Ack()
+	}
+	elapsed := time.Since(now)
+	if i != 0 {
+		t.Fatalf("Expected %d messages; got: %d", 0, i)
+	}
+	if msgs.Error() != nil {
+		t.Fatalf("Unexpected error: %s", msgs.Error())
+	}
+	if elapsed < 290*time.Millisecond {
+		t.Fatalf("Expected timeout after 300ms; got: %v", elapsed)
+	}
+
+	// delete consumer to verify heartbeats are not sent anymore
+	info, err := sub.ConsumerInfo()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if err := js.DeleteConsumer("TEST", info.Name); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	now = time.Now()
+	msgs, err = sub.FetchBatch(5, nats.PullHeartbeat(100*time.Millisecond), nats.MaxWait(1*time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	for msg := range msgs.Messages() {
+		msg.Ack()
+	}
+	elapsed = time.Since(now)
+	if elapsed < 200*time.Millisecond || elapsed > 300*time.Millisecond {
+		t.Fatalf("Expected timeout after 200ms and before 300ms; got: %v", elapsed)
+	}
+	if !errors.Is(msgs.Error(), nats.ErrNoHeartbeat) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+
+	// heartbeat value too large
+	_, err = sub.Fetch(5, nats.PullHeartbeat(200*time.Millisecond), nats.MaxWait(300*time.Millisecond))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+
+	// heartbeat value invalid
+	_, err = sub.Fetch(5, nats.PullHeartbeat(-1))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+}
+
 func TestPullSubscribeFetchBatch(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)