Repository: nats-io/nats.go
Pull request #1395 (2023-09-11 15:29:00)

feat: implement CreateOrUpdateStream function
After using JetStream on a private project I felt the need for CreateOrUpdateStream like CreateOrUpdateConsumer. This function updates if stream exists, otherwise it creates the stream. 


Patch:
diff --git a/.travis.yml b/.travis.yml
index 962109d..3687970 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,12 +1,12 @@
 language: go
 go:
+- "1.21.x"
 - "1.20.x"
-- "1.19.x"
 go_import_path: github.com/nats-io/nats.go
 install:
 - go get -t ./...
 - curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then
     go install github.com/mattn/goveralls@latest;
     go install github.com/wadey/gocovmerge@latest;
     go install honnef.co/go/tools/cmd/staticcheck@latest;
@@ -15,27 +15,22 @@ install:
 before_script:
 - $(exit $(go fmt ./... | wc -l))
 - go vet -modfile=go_test.mod ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then
     find . -type f -name "*.go" | xargs misspell -error -locale US;
     GOFLAGS="-mod=mod -modfile=go_test.mod" staticcheck ./...;
   fi
 - golangci-lint run ./jetstream/...
 script:
 - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
 after_success:
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.20 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi
 
 jobs:
   include:
-  - name: "Go: 1.20.x (nats-server@dev)"
-    go: "1.20.x"
-    before_script:
-    - go get -modfile go_test.mod github.com/nats-io/nats-server/v2@dev
-  - name: "Go: 1.20.x (nats-server@main)"
-    go: "1.20.x"
+  - name: "Go: 1.21.x (nats-server@main)"
+    go: "1.21.x"
     before_script:
     - go get -modfile go_test.mod github.com/nats-io/nats-server/v2@main
   allow_failures:
-  - name: "Go: 1.20.x (nats-server@dev)"
-  - name: "Go: 1.20.x (nats-server@main)"
+  - name: "Go: 1.21.x (nats-server@main)"
diff --git a/README.md b/README.md
index b38f98d..108db4e 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,7 @@ When using or transitioning to Go modules support:
 ```bash
 # Go client latest or explicit version
 go get github.com/nats-io/nats.go/@latest
-go get github.com/nats-io/nats.go/@v1.28.0
+go get github.com/nats-io/nats.go/@v1.30.2
 
 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
diff --git a/go.mod b/go.mod
index 10fd2bf..b103e07 100644
--- a/go.mod
+++ b/go.mod
@@ -1,12 +1,11 @@
 module github.com/nats-io/nats.go
 
-go 1.19
+go 1.20
 
 require (
-	github.com/klauspost/compress v1.16.5
-	github.com/nats-io/nkeys v0.4.4
+	github.com/klauspost/compress v1.17.0
+	github.com/nats-io/nkeys v0.4.5
 	github.com/nats-io/nuid v1.0.1
-	golang.org/x/text v0.7.0
 )
 
 require (
diff --git a/go.sum b/go.sum
index 92f2cca..111cfe2 100644
--- a/go.sum
+++ b/go.sum
@@ -1,12 +1,10 @@
-github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
-github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
-github.com/nats-io/nkeys v0.4.4 h1:xvBJ8d69TznjcQl9t6//Q5xXuVhyYiSos6RPtvQNTwA=
-github.com/nats-io/nkeys v0.4.4/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
+github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
+github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk=
+github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 golang.org/x/crypto v0.6.0 h1:qfktjS5LUO+fFKeJXZ+ikTRijMmljikvG68fpMMruSc=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
 golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
-golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
diff --git a/jetstream/README.md b/jetstream/README.md
index bd1be57..40b5962 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -253,16 +253,33 @@ CRUD operations on consumers can be achieved on 2 levels:
 js, _ := jetstream.New(nc)
 
 // create a consumer (this is an idempotent operation)
-cons, _ := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
+// an error will be returned if consumer already exists and has different configuration.
+cons, _ := js.CreateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
     Durable: "foo",
     AckPolicy: jetstream.AckExplicitPolicy,
 })
 
 // create an ephemeral pull consumer by not providing `Durable`
-ephemeral, _ := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
+ephemeral, _ := js.CreateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
     AckPolicy: jetstream.AckExplicitPolicy,
 })
 
+
+// consumer can also be created using CreateOrUpdateConsumer
+// this method will either create a consumer if it does not exist
+// or update existing consumer (if possible)
+cons2 := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
+    Name: "bar",
+})
+
+// consumers can be updated
+// an error will be returned if consumer with given name does not exists
+// or an illegal property is to be updated (e.g. AckPolicy)
+updated, _ := js.UpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
+    AckPolicy: jetstream.AckExplicitPolicy,
+    Description: "updated consumer"
+})
+
 // get consumer handle
 cons, _ = js.Consumer(ctx, "ORDERS", "foo")
 
@@ -280,7 +297,7 @@ js, _ := jetstream.New(nc)
 stream, _ := js.Stream(ctx, "ORDERS")
 
 // create consumer
-cons, _ := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
+cons, _ := stream.CreateConsumer(ctx, jetstream.ConsumerConfig{
     Durable:   "foo",
     AckPolicy: jetstream.AckExplicitPolicy,
 })
diff --git a/jetstream/consumer.go b/jetstream/consumer.go
index 05e0d6c..324327f 100644
--- a/jetstream/consumer.go
+++ b/jetstream/consumer.go
@@ -51,6 +51,12 @@ type (
 		// CachedInfo returns [*ConsumerInfo] cached on a consumer struct
 		CachedInfo() *ConsumerInfo
 	}
+
+	createConsumerRequest struct {
+		Stream string          `json:"stream_name"`
+		Config *ConsumerConfig `json:"config"`
+		Action string          `json:"action"`
+	}
 )
 
 // Info returns [ConsumerInfo] for a given consumer
@@ -71,6 +77,9 @@ func (p *pullConsumer) Info(ctx context.Context) (*ConsumerInfo, error) {
 		}
 		return nil, resp.Error
 	}
+	if resp.Error == nil && resp.ConsumerInfo == nil {
+		return nil, ErrConsumerNotFound
+	}
 
 	p.info = resp.ConsumerInfo
 	return resp.ConsumerInfo, nil
@@ -84,7 +93,7 @@ func (p *pullConsumer) CachedInfo() *ConsumerInfo {
 	return p.info
 }
 
-func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig) (Consumer, error) {
+func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string) (Consumer, error) {
 	ctx, cancel := wrapContextWithoutDeadline(ctx)
 	if cancel != nil {
 		defer cancel()
@@ -92,6 +101,7 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 	req := createConsumerRequest{
 		Stream: stream,
 		Config: &cfg,
+		Action: action,
 	}
 	reqJSON, err := json.Marshal(req)
 	if err != nil {
@@ -111,7 +121,7 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 	}
 
 	var ccSubj string
-	if cfg.FilterSubject != "" {
+	if cfg.FilterSubject != "" && len(cfg.FilterSubjects) == 0 {
 		ccSubj = apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject))
 	} else {
 		ccSubj = apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerCreateT, stream, consumerName))
@@ -128,6 +138,11 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 		return nil, resp.Error
 	}
 
+	// check whether multiple filter subjects (if used) are reflected in the returned ConsumerInfo
+	if len(cfg.FilterSubjects) != 0 && len(resp.Config.FilterSubjects) == 0 {
+		return nil, ErrConsumerMultipleFilterSubjectsNotSupported
+	}
+
 	return &pullConsumer{
 		jetStream:     js,
 		stream:        stream,
@@ -138,6 +153,12 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 	}, nil
 }
 
+const (
+	consumerActionCreate         = "create"
+	consumerActionUpdate         = "update"
+	consumerActionCreateOrUpdate = ""
+)
+
 func generateConsName() string {
 	name := nuid.Next()
 	sha := sha256.New()
@@ -170,6 +191,9 @@ func getConsumer(ctx context.Context, js *jetStream, stream, name string) (Consu
 		}
 		return nil, resp.Error
 	}
+	if resp.Error == nil && resp.ConsumerInfo == nil {
+		return nil, ErrConsumerNotFound
+	}
 
 	cons := &pullConsumer{
 		jetStream:     js,
diff --git a/jetstream/consumer_config.go b/jetstream/consumer_config.go
index 41764ba..6729b47 100644
--- a/jetstream/consumer_config.go
+++ b/jetstream/consumer_config.go
@@ -69,6 +69,11 @@ type (
 
 		// NOTE: FilterSubjects requires nats-server v2.10.0+
 		FilterSubjects []string `json:"filter_subjects,omitempty"`
+
+		// Metadata is additional metadata for the Consumer.
+		// Keys starting with `_nats` are reserved.
+		// NOTE: Metadata requires nats-server v2.10.0+
+		Metadata map[string]string `json:"metadata,omitempty"`
 	}
 
 	OrderedConsumerConfig struct {
diff --git a/jetstream/errors.go b/jetstream/errors.go
index ffefea5..e8d1a77 100644
--- a/jetstream/errors.go
+++ b/jetstream/errors.go
@@ -49,10 +49,15 @@ const (
 	JSErrCodeStreamNotFound  ErrorCode = 10059
 	JSErrCodeStreamNameInUse ErrorCode = 10058
 
-	JSErrCodeConsumerCreate        ErrorCode = 10012
-	JSErrCodeConsumerNotFound      ErrorCode = 10014
-	JSErrCodeConsumerNameExists    ErrorCode = 10013
-	JSErrCodeConsumerAlreadyExists ErrorCode = 10105
+	JSErrCodeConsumerCreate            ErrorCode = 10012
+	JSErrCodeConsumerNotFound          ErrorCode = 10014
+	JSErrCodeConsumerNameExists        ErrorCode = 10013
+	JSErrCodeConsumerAlreadyExists     ErrorCode = 10105
+	JSErrCodeConsumerExists            ErrorCode = 10148
+	JSErrCodeDuplicateFilterSubjects   ErrorCode = 10136
+	JSErrCodeOverlappingFilterSubjects ErrorCode = 10138
+	JSErrCodeConsumerEmptyFilter       ErrorCode = 10139
+	JSErrCodeConsumerDoesNotExist      ErrorCode = 10149
 
 	JSErrCodeMessageNotFound ErrorCode = 10037
 
@@ -76,9 +81,35 @@ var (
 	// ErrStreamNameAlreadyInUse is returned when a stream with given name already exists and has a different configuration.
 	ErrStreamNameAlreadyInUse JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNameInUse, Description: "stream name already in use", Code: 400}}
 
+	// ErrStreamSubjectTransformNotSupported is returned when the connected nats-server version does not support setting
+	// the stream subject transform. If this error is returned when executing CreateStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSubjectTransformNotSupported JetStreamError = &jsError{message: "stream subject transformation not supported by nats-server"}
+
+	// ErrStreamSourceSubjectTransformNotSupported is returned when the connected nats-server version does not support setting
+	// the stream source subject transform. If this error is returned when executing CreateStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSourceSubjectTransformNotSupported JetStreamError = &jsError{message: "stream subject transformation not supported by nats-server"}
+
+	// ErrStreamSourceNotSupported is returned when the connected nats-server version does not support setting
+	// the stream sources. If this error is returned when executing CreateStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSourceNotSupported JetStreamError = &jsError{message: "stream sourcing is not supported by nats-server"}
+
+	// ErrStreamSourceMultipleFilterSubjectsNotSupported is returned when the connected nats-server version does not support setting
+	// the stream sources. If this error is returned when executing CreateStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSourceMultipleFilterSubjectsNotSupported JetStreamError = &jsError{message: "stream sourcing with multiple subject filters not supported by nats-server"}
+
 	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
 	ErrConsumerNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerNotFound, Description: "consumer not found", Code: 404}}
 
+	// ErrConsumerExists is returned when attempting to create a consumer with CreateConsumer but a consumer with given name already exists.
+	ErrConsumerExists JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerExists, Description: "consumer already exists", Code: 400}}
+
+	// ErrConsumerNameExists is returned when attempting to update a consumer with UpdateConsumer but a consumer with given name does not exist.
+	ErrConsumerDoesNotExist JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerDoesNotExist, Description: "consumer does not exist", Code: 400}}
+
 	// ErrMsgNotFound is returned when message with provided sequence number does not exist.
 	ErrMsgNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeMessageNotFound, Description: "message not found", Code: 404}}
 
@@ -88,8 +119,22 @@ var (
 	// ErrConsumerCreate is returned when nats-server reports error when creating consumer (e.g. illegal update).
 	ErrConsumerCreate JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerCreate, Description: "could not create consumer", Code: 500}}
 
+	// ErrDuplicateFilterSubjects is returned when both FilterSubject and FilterSubjects are specified when creating consumer.
+	ErrDuplicateFilterSubjects JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeDuplicateFilterSubjects, Description: "consumer cannot have both FilterSubject and FilterSubjects specified", Code: 500}}
+
+	// ErrDuplicateFilterSubjects is returned when filter subjects overlap when creating consumer.
+	ErrOverlappingFilterSubjects JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeOverlappingFilterSubjects, Description: "consumer subject filters cannot overlap", Code: 500}}
+
+	// ErrEmptyFilter is returned when a filter in FilterSubjects is empty.
+	ErrEmptyFilter JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerEmptyFilter, Description: "consumer filter in FilterSubjects cannot be empty", Code: 500}}
+
 	// Client errors
 
+	// ErrConsumerMultipleFilterSubjectsNotSupported is returned when the connected nats-server version does not support setting
+	// multiple filter subjects with filter_subjects field. If this error is returned when executing AddConsumer(), the consumer with invalid
+	// configuration was already created in the server.
+	ErrConsumerMultipleFilterSubjectsNotSupported JetStreamError = &jsError{message: "multiple consumer filter subjects not supported by nats-server"}
+
 	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
 	ErrConsumerNameAlreadyInUse JetStreamError = &jsError{message: "consumer name already in use"}
 
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 1ff86c3..927ac46 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -69,6 +69,8 @@ type (
 		CreateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
 		// UpdateStream updates an existing stream
 		UpdateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
+		// CreateOrUpdateStream creates a stream with given config. If stream already exists, it will be updated (if possible).
+		CreateOrUpdateStream(ctx context.Context, cfg StreamConfig) (Stream, error)
 		// Stream returns a [Stream] hook for a given stream name
 		Stream(ctx context.Context, stream string) (Stream, error)
 		// StreamNameBySubject returns a stream name stream listening on given subject
@@ -86,6 +88,14 @@ type (
 		// If consumer already exists, it will be updated (if possible).
 		// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages)
 		CreateOrUpdateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error)
+		// CreateConsumer creates a consumer on a given stream with given config.
+		// If consumer already exists, ErrConsumerExists is returned.
+		// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages)
+		CreateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error)
+		// UpdateConsumer updates an existing consumer.
+		// If consumer does not exist, ErrConsumerDoesNotExist is returned.
+		// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages)
+		UpdateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error)
 		// OrderedConsumer returns an OrderedConsumer instance.
 		// OrderedConsumer allows fetching messages from a stream (just like standard consumer),
 		// for in order delivery of messages. Underlying consumer is re-created when necessary,
@@ -350,6 +360,22 @@ func (js *jetStream) CreateStream(ctx context.Context, cfg StreamConfig) (Stream
 		return nil, resp.Error
 	}
 
+	// check that input subject transform (if used) is reflected in the returned StreamInfo
+	if cfg.SubjectTransform != nil && resp.StreamInfo.Config.SubjectTransform == nil {
+		return nil, ErrStreamSubjectTransformNotSupported
+	}
+
+	if len(cfg.Sources) != 0 {
+		if len(cfg.Sources) != len(resp.Config.Sources) {
+			return nil, ErrStreamSourceNotSupported
+		}
+		for i := range cfg.Sources {
+			if len(cfg.Sources[i].SubjectTransforms) != 0 && len(resp.Sources[i].SubjectTransforms) == 0 {
+				return nil, ErrStreamSourceMultipleFilterSubjectsNotSupported
+			}
+		}
+	}
+
 	return &stream{
 		jetStream: js,
 		name:      cfg.Name,
@@ -413,6 +439,22 @@ func (js *jetStream) UpdateStream(ctx context.Context, cfg StreamConfig) (Stream
 		return nil, resp.Error
 	}
 
+	// check that input subject transform (if used) is reflected in the returned StreamInfo
+	if cfg.SubjectTransform != nil && resp.StreamInfo.Config.SubjectTransform == nil {
+		return nil, ErrStreamSubjectTransformNotSupported
+	}
+
+	if len(cfg.Sources) != 0 {
+		if len(cfg.Sources) != len(resp.Config.Sources) {
+			return nil, ErrStreamSourceNotSupported
+		}
+		for i := range cfg.Sources {
+			if len(cfg.Sources[i].SubjectTransforms) != 0 && len(resp.Sources[i].SubjectTransforms) == 0 {
+				return nil, ErrStreamSourceMultipleFilterSubjectsNotSupported
+			}
+		}
+	}
+
 	return &stream{
 		jetStream: js,
 		name:      cfg.Name,
@@ -420,6 +462,18 @@ func (js *jetStream) UpdateStream(ctx context.Context, cfg StreamConfig) (Stream
 	}, nil
 }
 
+func (js *jetStream) CreateOrUpdateStream(ctx context.Context, cfg StreamConfig) (Stream, error) {
+	s, err := js.UpdateStream(ctx, cfg)
+	if err != nil {
+		if !errors.Is(err, ErrStreamNotFound) {
+			return nil, err
+		}
+		return js.CreateStream(ctx, cfg)
+	}
+
+	return s, nil
+}
+
 // Stream returns a [Stream] hook for a given stream name
 func (js *jetStream) Stream(ctx context.Context, name string) (Stream, error) {
 	if err := validateStreamName(name); err != nil {
@@ -480,7 +534,21 @@ func (js *jetStream) CreateOrUpdateConsumer(ctx context.Context, stream string,
 	if err := validateStreamName(stream); err != nil {
 		return nil, err
 	}
-	return upsertConsumer(ctx, js, stream, cfg)
+	return upsertConsumer(ctx, js, stream, cfg, consumerActionCreateOrUpdate)
+}
+
+func (js *jetStream) CreateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error) {
+	if err := validateStreamName(stream); err != nil {
+		return nil, err
+	}
+	return upsertConsumer(ctx, js, stream, cfg, consumerActionCreate)
+}
+
+func (js *jetStream) UpdateConsumer(ctx context.Context, stream string, cfg ConsumerConfig) (Consumer, error) {
+	if err := validateStreamName(stream); err != nil {
+		return nil, err
+	}
+	return upsertConsumer(ctx, js, stream, cfg, consumerActionUpdate)
 }
 
 func (js *jetStream) OrderedConsumer(ctx context.Context, stream string, cfg OrderedConsumerConfig) (Consumer, error) {
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 3546f01..c16c571 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -351,12 +351,20 @@ func (js *jetStream) CreateKeyValue(ctx context.Context, cfg KeyValueConfig) (Ke
 	} else if len(cfg.Sources) > 0 {
 		// For now we do not allow direct subjects for sources. If that is desired a user could use stream API directly.
 		for _, ss := range cfg.Sources {
-			if !strings.HasPrefix(ss.Name, kvBucketNamePre) {
-				ss = ss.copy()
+			var sourceBucketName string
+			if strings.HasPrefix(ss.Name, kvBucketNamePre) {
+				sourceBucketName = ss.Name[len(kvBucketNamePre):]
+			} else {
+				sourceBucketName = ss.Name
 				ss.Name = fmt.Sprintf(kvBucketNameTmpl, ss.Name)
 			}
+
+			if ss.External == nil || sourceBucketName != cfg.Bucket {
+				ss.SubjectTransforms = []SubjectTransformConfig{{Source: fmt.Sprintf(kvSubjectsTmpl, sourceBucketName), Destination: fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}}
+			}
 			scfg.Sources = append(scfg.Sources, ss)
 		}
+		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
 	} else {
 		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
 	}
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..e462981 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()
@@ -447,6 +519,9 @@ func (c *orderedConsumer) Info(ctx context.Context) (*ConsumerInfo, error) {
 		}
 		return nil, resp.Error
 	}
+	if resp.Error == nil && resp.ConsumerInfo == nil {
+		return nil, ErrConsumerNotFound
+	}
 
 	c.currentConsumer.info = resp.ConsumerInfo
 	return resp.ConsumerInfo, nil
diff --git a/jetstream/publish.go b/jetstream/publish.go
index 1215687..30371cd 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -402,6 +402,10 @@ func (js *jetStream) resetPendingAcksOnReconnect() {
 			paf.err = nats.ErrDisconnected
 		}
 		js.publisher.acks = nil
+		if js.publisher.doneCh != nil {
+			close(js.publisher.doneCh)
+			js.publisher.doneCh = nil
+		}
 		js.publisher.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/jetstream/stream.go b/jetstream/stream.go
index dfb5244..e864cf1 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -56,6 +56,16 @@ type (
 		// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages).
 		CreateOrUpdateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error)
 
+		// CreateConsumer creates a consumer on a given stream with given config.
+		// If consumer already exists, an ErrConsumerExists is returned.
+		// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages).
+		CreateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error)
+
+		// UpdateConsumer updates an existing consumer with given config.
+		// If consumer does not exist, an ErrConsumerDoesNotExist is returned.
+		// Consumer interface is returned, serving as a hook to operate on a consumer (e.g. fetch messages).
+		UpdateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error)
+
 		// OrderedConsumer returns an OrderedConsumer instance.
 		// OrderedConsumer allows fetching messages from a stream (just like standard consumer),
 		// for in order delivery of messages. Underlying consumer is re-created when necessary,
@@ -74,6 +84,7 @@ type (
 		// ConsumerNames returns a ConsumerNameLister enabling iterating over a channel of consumer names
 		ConsumerNames(context.Context) ConsumerNameLister
 	}
+
 	RawStreamMsg struct {
 		Subject  string
 		Sequence uint64
@@ -100,11 +111,6 @@ type (
 		*ConsumerInfo
 	}
 
-	createConsumerRequest struct {
-		Stream string          `json:"stream_name"`
-		Config *ConsumerConfig `json:"config"`
-	}
-
 	StreamPurgeOpt func(*StreamPurgeRequest) error
 
 	StreamPurgeRequest struct {
@@ -194,7 +200,15 @@ type (
 )
 
 func (s *stream) CreateOrUpdateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error) {
-	return upsertConsumer(ctx, s.jetStream, s.name, cfg)
+	return upsertConsumer(ctx, s.jetStream, s.name, cfg, consumerActionCreateOrUpdate)
+}
+
+func (s *stream) CreateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error) {
+	return upsertConsumer(ctx, s.jetStream, s.name, cfg, consumerActionCreate)
+}
+
+func (s *stream) UpdateConsumer(ctx context.Context, cfg ConsumerConfig) (Consumer, error) {
+	return upsertConsumer(ctx, s.jetStream, s.name, cfg, consumerActionUpdate)
 }
 
 func (s *stream) OrderedConsumer(ctx context.Context, cfg OrderedConsumerConfig) (Consumer, error) {
diff --git a/jetstream/stream_config.go b/jetstream/stream_config.go
index 9b5461d..bb4fd10 100644
--- a/jetstream/stream_config.go
+++ b/jetstream/stream_config.go
@@ -35,30 +35,35 @@ type (
 	}
 
 	StreamConfig struct {
-		Name                 string          `json:"name"`
-		Description          string          `json:"description,omitempty"`
-		Subjects             []string        `json:"subjects,omitempty"`
-		Retention            RetentionPolicy `json:"retention"`
-		MaxConsumers         int             `json:"max_consumers"`
-		MaxMsgs              int64           `json:"max_msgs"`
-		MaxBytes             int64           `json:"max_bytes"`
-		Discard              DiscardPolicy   `json:"discard"`
-		DiscardNewPerSubject bool            `json:"discard_new_per_subject,omitempty"`
-		MaxAge               time.Duration   `json:"max_age"`
-		MaxMsgsPerSubject    int64           `json:"max_msgs_per_subject"`
-		MaxMsgSize           int32           `json:"max_msg_size,omitempty"`
-		Storage              StorageType     `json:"storage"`
-		Replicas             int             `json:"num_replicas"`
-		NoAck                bool            `json:"no_ack,omitempty"`
-		Template             string          `json:"template_owner,omitempty"`
-		Duplicates           time.Duration   `json:"duplicate_window,omitempty"`
-		Placement            *Placement      `json:"placement,omitempty"`
-		Mirror               *StreamSource   `json:"mirror,omitempty"`
-		Sources              []*StreamSource `json:"sources,omitempty"`
-		Sealed               bool            `json:"sealed,omitempty"`
-		DenyDelete           bool            `json:"deny_delete,omitempty"`
-		DenyPurge            bool            `json:"deny_purge,omitempty"`
-		AllowRollup          bool            `json:"allow_rollup_hdrs,omitempty"`
+		Name                 string           `json:"name"`
+		Description          string           `json:"description,omitempty"`
+		Subjects             []string         `json:"subjects,omitempty"`
+		Retention            RetentionPolicy  `json:"retention"`
+		MaxConsumers         int              `json:"max_consumers"`
+		MaxMsgs              int64            `json:"max_msgs"`
+		MaxBytes             int64            `json:"max_bytes"`
+		Discard              DiscardPolicy    `json:"discard"`
+		DiscardNewPerSubject bool             `json:"discard_new_per_subject,omitempty"`
+		MaxAge               time.Duration    `json:"max_age"`
+		MaxMsgsPerSubject    int64            `json:"max_msgs_per_subject"`
+		MaxMsgSize           int32            `json:"max_msg_size,omitempty"`
+		Storage              StorageType      `json:"storage"`
+		Replicas             int              `json:"num_replicas"`
+		NoAck                bool             `json:"no_ack,omitempty"`
+		Template             string           `json:"template_owner,omitempty"`
+		Duplicates           time.Duration    `json:"duplicate_window,omitempty"`
+		Placement            *Placement       `json:"placement,omitempty"`
+		Mirror               *StreamSource    `json:"mirror,omitempty"`
+		Sources              []*StreamSource  `json:"sources,omitempty"`
+		Sealed               bool             `json:"sealed,omitempty"`
+		DenyDelete           bool             `json:"deny_delete,omitempty"`
+		DenyPurge            bool             `json:"deny_purge,omitempty"`
+		AllowRollup          bool             `json:"allow_rollup_hdrs,omitempty"`
+		Compression          StoreCompression `json:"compression"`
+		FirstSeq             uint64           `json:"first_seq,omitempty"`
+
+		// Allow applying a subject transform to incoming messages before doing anything else
+		SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`
 
 		// Allow republish of the message after being sequenced and stored.
 		RePublish *RePublish `json:"republish,omitempty"`
@@ -67,13 +72,23 @@ type (
 		AllowDirect bool `json:"allow_direct"`
 		// Allow higher performance and unified direct access for mirrors as well.
 		MirrorDirect bool `json:"mirror_direct"`
+
+		// Limits for consumers on this stream.
+		ConsumerLimits StreamConsumerLimits `json:"consumer_limits,omitempty"`
+
+		// Metadata is additional metadata for the Stream.
+		// Keys starting with `_nats` are reserved.
+		// NOTE: Metadata requires nats-server v2.10.0+
+		Metadata map[string]string `json:"metadata,omitempty"`
 	}
 
 	// StreamSourceInfo shows information about an upstream stream source.
 	StreamSourceInfo struct {
-		Name   string        `json:"name"`
-		Lag    uint64        `json:"lag"`
-		Active time.Duration `json:"active"`
+		Name              string                   `json:"name"`
+		Lag               uint64                   `json:"lag"`
+		Active            time.Duration            `json:"active"`
+		FilterSubject     string                   `json:"filter_subject,omitempty"`
+		SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
 	}
 
 	// StreamState is information about the given stream.
@@ -109,6 +124,12 @@ type (
 		Lag     uint64        `json:"lag,omitempty"`
 	}
 
+	// SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received
+	SubjectTransformConfig struct {
+		Source      string `json:"src"`
+		Destination string `json:"dest"`
+	}
+
 	// RePublish is for republishing messages once committed to a stream. The original
 	// subject is remapped from the subject pattern to the destination pattern.
 	RePublish struct {
@@ -125,12 +146,13 @@ type (
 
 	// StreamSource dictates how streams can source from other streams.
 	StreamSource struct {
-		Name          string          `json:"name"`
-		OptStartSeq   uint64          `json:"opt_start_seq,omitempty"`
-		OptStartTime  *time.Time      `json:"opt_start_time,omitempty"`
-		FilterSubject string          `json:"filter_subject,omitempty"`
-		External      *ExternalStream `json:"external,omitempty"`
-		Domain        string          `json:"-"`
+		Name              string                   `json:"name"`
+		OptStartSeq       uint64                   `json:"opt_start_seq,omitempty"`
+		OptStartTime      *time.Time               `json:"opt_start_time,omitempty"`
+		FilterSubject     string                   `json:"filter_subject,omitempty"`
+		SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
+		External          *ExternalStream          `json:"external,omitempty"`
+		Domain            string                   `json:"-"`
 	}
 
 	// ExternalStream allows you to qualify access to a stream source in another
@@ -140,6 +162,13 @@ type (
 		DeliverPrefix string `json:"deliver"`
 	}
 
+	// StreamConsumerLimits are the limits for a consumer on a stream.
+	// These can be overridden on a per consumer basis.
+	StreamConsumerLimits struct {
+		InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
+		MaxAckPending     int           `json:"max_ack_pending,omitempty"`
+	}
+
 	// DiscardPolicy determines how to proceed when limits of messages or bytes are
 	// reached.
 	DiscardPolicy int
@@ -149,6 +178,9 @@ type (
 
 	// StorageType determines how messages are stored for retention.
 	StorageType int
+
+	// StoreCompression determines how messages are compressed.
+	StoreCompression uint8
 )
 
 const (
@@ -299,3 +331,48 @@ func (st *StorageType) UnmarshalJSON(data []byte) error {
 func jsonString(s string) string {
 	return "\"" + s + "\""
 }
+
+const (
+	NoCompression StoreCompression = iota
+	S2Compression
+)
+
+func (alg StoreCompression) String() string {
+	switch alg {
+	case NoCompression:
+		return "None"
+	case S2Compression:
+		return "S2"
+	default:
+		return "Unknown StoreCompression"
+	}
+}
+
+func (alg StoreCompression) MarshalJSON() ([]byte, error) {
+	var str string
+	switch alg {
+	case S2Compression:
+		str = "s2"
+	case NoCompression:
+		str = "none"
+	default:
+		return nil, fmt.Errorf("unknown compression algorithm")
+	}
+	return json.Marshal(str)
+}
+
+func (alg *StoreCompression) UnmarshalJSON(b []byte) error {
+	var str string
+	if err := json.Unmarshal(b, &str); err != nil {
+		return err
+	}
+	switch str {
+	case "s2":
+		*alg = S2Compression
+	case "none":
+		*alg = NoCompression
+	default:
+		return fmt.Errorf("unknown compression algorithm")
+	}
+	return nil
+}
diff --git a/js.go b/js.go
index 05637d5..444278e 100644
--- a/js.go
+++ b/js.go
@@ -698,6 +698,10 @@ func (js *js) resetPendingAcksOnReconnect() {
 			paf.err = ErrDisconnected
 		}
 		js.pafs = nil
+		if js.dch != nil {
+			close(js.dch)
+			js.dch = nil
+		}
 		js.mu.Unlock()
 	}
 }
@@ -1114,6 +1118,7 @@ type ConsumerConfig struct {
 	MaxDeliver      int             `json:"max_deliver,omitempty"`
 	BackOff         []time.Duration `json:"backoff,omitempty"`
 	FilterSubject   string          `json:"filter_subject,omitempty"`
+	FilterSubjects  []string        `json:"filter_subjects,omitempty"`
 	ReplayPolicy    ReplayPolicy    `json:"replay_policy"`
 	RateLimit       uint64          `json:"rate_limit_bps,omitempty"` // Bits per sec
 	SampleFrequency string          `json:"sample_freq,omitempty"`
@@ -1139,6 +1144,11 @@ type ConsumerConfig struct {
 	Replicas int `json:"num_replicas"`
 	// Force memory storage.
 	MemoryStorage bool `json:"mem_storage,omitempty"`
+
+	// Metadata is additional metadata for the Consumer.
+	// Keys starting with `_nats` are reserved.
+	// NOTE: Metadata requires nats-server v2.10.0+
+	Metadata map[string]string `json:"metadata,omitempty"`
 }
 
 // ConsumerInfo is the info from a JetStream consumer.
@@ -1172,10 +1182,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.
@@ -2465,6 +2476,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
@@ -2564,6 +2576,16 @@ func ConsumerName(name string) SubOpt {
 	})
 }
 
+// ConsumerFilterSubjects can be used to set multiple subject filters on the consumer.
+// It has to be used in conjunction with [nats.BindStream] and
+// with empty 'subject' parameter.
+func ConsumerFilterSubjects(subjects ...string) SubOpt {
+	return subOptFn(func(opts *subOpts) error {
+		opts.cfg.FilterSubjects = subjects
+		return nil
+	})
+}
+
 func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
 	sub.mu.Lock()
 	// TODO(dlc) - Better way to mark especially if we attach.
@@ -2584,6 +2606,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.
@@ -2599,6 +2622,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
 
@@ -2642,6 +2675,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
@@ -2728,7 +2766,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.
@@ -2739,7 +2776,17 @@ 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()
+
+	// if heartbeat is set, validate it against the context timeout
+	if o.hb > 0 {
+		deadline, _ := ctx.Deadline()
+		if 2*o.hb >= time.Until(deadline) {
+			return nil, fmt.Errorf("%w: idle heartbeat value too large", ErrInvalidArg)
+		}
 	}
 
 	// Check if context not done already before making the request.
@@ -2779,6 +2826,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.
@@ -2809,8 +2858,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()
@@ -2818,6 +2885,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)
@@ -2836,9 +2906,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
@@ -2966,14 +3042,24 @@ 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()
 		}
 	}()
 
+	// if heartbeat is set, validate it against the context timeout
+	if o.hb > 0 {
+		deadline, _ := ctx.Deadline()
+		if 2*o.hb >= time.Until(deadline) {
+			return nil, fmt.Errorf("%w: idle heartbeat value too large", ErrInvalidArg)
+		}
+	}
+
 	// Check if context not done already before making the request.
 	select {
 	case <-ctx.Done():
@@ -3027,9 +3113,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 {
@@ -3047,11 +3134,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
@@ -3059,6 +3152,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)
@@ -3078,7 +3174,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{}{}
@@ -3123,6 +3223,9 @@ func (js *js) getConsumerInfoContext(ctx context.Context, stream, consumer strin
 		}
 		return nil, info.Error
 	}
+	if info.Error == nil && info.ConsumerInfo == nil {
+		return nil, ErrConsumerNotFound
+	}
 	return info.ConsumerInfo, nil
 }
 
@@ -3650,6 +3753,53 @@ func (st *StorageType) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+type StoreCompression uint8
+
+const (
+	NoCompression StoreCompression = iota
+	S2Compression
+)
+
+func (alg StoreCompression) String() string {
+	switch alg {
+	case NoCompression:
+		return "None"
+	case S2Compression:
+		return "S2"
+	default:
+		return "Unknown StoreCompression"
+	}
+}
+
+func (alg StoreCompression) MarshalJSON() ([]byte, error) {
+	var str string
+	switch alg {
+	case S2Compression:
+		str = "s2"
+	case NoCompression:
+		str = "none"
+	default:
+		return nil, fmt.Errorf("unknown compression algorithm")
+	}
+	return json.Marshal(str)
+}
+
+func (alg *StoreCompression) UnmarshalJSON(b []byte) error {
+	var str string
+	if err := json.Unmarshal(b, &str); err != nil {
+		return err
+	}
+	switch str {
+	case "s2":
+		*alg = S2Compression
+	case "none":
+		*alg = NoCompression
+	default:
+		return fmt.Errorf("unknown compression algorithm")
+	}
+	return nil
+}
+
 // Length of our hash used for named consumers.
 const nameHashLen = 8
 
diff --git a/jserrors.go b/jserrors.go
index d7959ca..c8b1f5f 100644
--- a/jserrors.go
+++ b/jserrors.go
@@ -33,6 +33,26 @@ var (
 	// ErrStreamNameAlreadyInUse is returned when a stream with given name already exists and has a different configuration.
 	ErrStreamNameAlreadyInUse JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNameInUse, Description: "stream name already in use", Code: 400}}
 
+	// ErrStreamSubjectTransformNotSupported is returned when the connected nats-server version does not support setting
+	// the stream subject transform. If this error is returned when executing AddStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSubjectTransformNotSupported JetStreamError = &jsError{message: "stream subject transformation not supported by nats-server"}
+
+	// ErrStreamSourceSubjectTransformNotSupported is returned when the connected nats-server version does not support setting
+	// the stream source subject transform. If this error is returned when executing AddStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSourceSubjectTransformNotSupported JetStreamError = &jsError{message: "stream subject transformation not supported by nats-server"}
+
+	// ErrStreamSourceNotSupported is returned when the connected nats-server version does not support setting
+	// the stream sources. If this error is returned when executing AddStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSourceNotSupported JetStreamError = &jsError{message: "stream sourcing is not supported by nats-server"}
+
+	// ErrStreamSourceMultipleSubjectTransformsNotSupported is returned when the connected nats-server version does not support setting
+	// the stream sources. If this error is returned when executing AddStream(), the stream with invalid
+	// configuration was already created in the server.
+	ErrStreamSourceMultipleSubjectTransformsNotSupported JetStreamError = &jsError{message: "stream sourceing with multiple subject transforms not supported by nats-server"}
+
 	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
 	ErrConsumerNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerNotFound, Description: "consumer not found", Code: 404}}
 
@@ -42,6 +62,15 @@ var (
 	// ErrBadRequest is returned when invalid request is sent to JetStream API.
 	ErrBadRequest JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeBadRequest, Description: "bad request", Code: 400}}
 
+	// ErrDuplicateFilterSubjects is returned when both FilterSubject and FilterSubjects are specified when creating consumer.
+	ErrDuplicateFilterSubjects JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeDuplicateFilterSubjects, Description: "consumer cannot have both FilterSubject and FilterSubjects specified", Code: 500}}
+
+	// ErrDuplicateFilterSubjects is returned when filter subjects overlap when creating consumer.
+	ErrOverlappingFilterSubjects JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeOverlappingFilterSubjects, Description: "consumer subject filters cannot overlap", Code: 500}}
+
+	// ErrEmptyFilter is returned when a filter in FilterSubjects is empty.
+	ErrEmptyFilter JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerEmptyFilter, Description: "consumer filter in FilterSubjects cannot be empty", Code: 500}}
+
 	// Client errors
 
 	// ErrConsumerNameAlreadyInUse is an error returned when consumer with given name already exists.
@@ -62,6 +91,11 @@ var (
 	// ErrConsumerNameRequired is returned when the provided consumer durable name is empty.
 	ErrConsumerNameRequired JetStreamError = &jsError{message: "consumer name is required"}
 
+	// ErrConsumerMultipleFilterSubjectsNotSupported is returned when the connected nats-server version does not support setting
+	// multiple filter subjects with filter_subjects field. If this error is returned when executing AddConsumer(), the consumer with invalid
+	// configuration was already created in the server.
+	ErrConsumerMultipleFilterSubjectsNotSupported JetStreamError = &jsError{message: "multiple consumer filter subjects not supported by nats-server"}
+
 	// ErrConsumerConfigRequired is returned when empty consumer consuguration is supplied to add/update consumer.
 	ErrConsumerConfigRequired JetStreamError = &jsError{message: "consumer configuration is required"}
 
@@ -104,6 +138,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")
@@ -120,13 +157,17 @@ const (
 	JSErrCodeStreamNotFound  ErrorCode = 10059
 	JSErrCodeStreamNameInUse ErrorCode = 10058
 
-	JSErrCodeConsumerNotFound      ErrorCode = 10014
-	JSErrCodeConsumerNameExists    ErrorCode = 10013
-	JSErrCodeConsumerAlreadyExists ErrorCode = 10105
+	JSErrCodeConsumerNotFound          ErrorCode = 10014
+	JSErrCodeConsumerNameExists        ErrorCode = 10013
+	JSErrCodeConsumerAlreadyExists     ErrorCode = 10105
+	JSErrCodeDuplicateFilterSubjects   ErrorCode = 10136
+	JSErrCodeOverlappingFilterSubjects ErrorCode = 10138
+	JSErrCodeConsumerEmptyFilter       ErrorCode = 10139
 
 	JSErrCodeMessageNotFound ErrorCode = 10037
 
-	JSErrCodeBadRequest ErrorCode = 10003
+	JSErrCodeBadRequest   ErrorCode = 10003
+	JSStreamInvalidConfig ErrorCode = 10052
 
 	JSErrCodeStreamWrongLastSequence ErrorCode = 10071
 )
diff --git a/jsm.go b/jsm.go
index c668469..266bf06 100644
--- a/jsm.go
+++ b/jsm.go
@@ -102,30 +102,35 @@ type JetStreamManager interface {
 // There are sensible defaults for most. If no subjects are
 // given the name will be used as the only subject.
 type StreamConfig struct {
-	Name                 string          `json:"name"`
-	Description          string          `json:"description,omitempty"`
-	Subjects             []string        `json:"subjects,omitempty"`
-	Retention            RetentionPolicy `json:"retention"`
-	MaxConsumers         int             `json:"max_consumers"`
-	MaxMsgs              int64           `json:"max_msgs"`
-	MaxBytes             int64           `json:"max_bytes"`
-	Discard              DiscardPolicy   `json:"discard"`
-	DiscardNewPerSubject bool            `json:"discard_new_per_subject,omitempty"`
-	MaxAge               time.Duration   `json:"max_age"`
-	MaxMsgsPerSubject    int64           `json:"max_msgs_per_subject"`
-	MaxMsgSize           int32           `json:"max_msg_size,omitempty"`
-	Storage              StorageType     `json:"storage"`
-	Replicas             int             `json:"num_replicas"`
-	NoAck                bool            `json:"no_ack,omitempty"`
-	Template             string          `json:"template_owner,omitempty"`
-	Duplicates           time.Duration   `json:"duplicate_window,omitempty"`
-	Placement            *Placement      `json:"placement,omitempty"`
-	Mirror               *StreamSource   `json:"mirror,omitempty"`
-	Sources              []*StreamSource `json:"sources,omitempty"`
-	Sealed               bool            `json:"sealed,omitempty"`
-	DenyDelete           bool            `json:"deny_delete,omitempty"`
-	DenyPurge            bool            `json:"deny_purge,omitempty"`
-	AllowRollup          bool            `json:"allow_rollup_hdrs,omitempty"`
+	Name                 string           `json:"name"`
+	Description          string           `json:"description,omitempty"`
+	Subjects             []string         `json:"subjects,omitempty"`
+	Retention            RetentionPolicy  `json:"retention"`
+	MaxConsumers         int              `json:"max_consumers"`
+	MaxMsgs              int64            `json:"max_msgs"`
+	MaxBytes             int64            `json:"max_bytes"`
+	Discard              DiscardPolicy    `json:"discard"`
+	DiscardNewPerSubject bool             `json:"discard_new_per_subject,omitempty"`
+	MaxAge               time.Duration    `json:"max_age"`
+	MaxMsgsPerSubject    int64            `json:"max_msgs_per_subject"`
+	MaxMsgSize           int32            `json:"max_msg_size,omitempty"`
+	Storage              StorageType      `json:"storage"`
+	Replicas             int              `json:"num_replicas"`
+	NoAck                bool             `json:"no_ack,omitempty"`
+	Template             string           `json:"template_owner,omitempty"`
+	Duplicates           time.Duration    `json:"duplicate_window,omitempty"`
+	Placement            *Placement       `json:"placement,omitempty"`
+	Mirror               *StreamSource    `json:"mirror,omitempty"`
+	Sources              []*StreamSource  `json:"sources,omitempty"`
+	Sealed               bool             `json:"sealed,omitempty"`
+	DenyDelete           bool             `json:"deny_delete,omitempty"`
+	DenyPurge            bool             `json:"deny_purge,omitempty"`
+	AllowRollup          bool             `json:"allow_rollup_hdrs,omitempty"`
+	Compression          StoreCompression `json:"compression"`
+	FirstSeq             uint64           `json:"first_seq,omitempty"`
+
+	// Allow applying a subject transform to incoming messages before doing anything else.
+	SubjectTransform *SubjectTransformConfig `json:"subject_transform,omitempty"`
 
 	// Allow republish of the message after being sequenced and stored.
 	RePublish *RePublish `json:"republish,omitempty"`
@@ -134,6 +139,20 @@ type StreamConfig struct {
 	AllowDirect bool `json:"allow_direct"`
 	// Allow higher performance and unified direct access for mirrors as well.
 	MirrorDirect bool `json:"mirror_direct"`
+
+	// Limits for consumers on this stream.
+	ConsumerLimits StreamConsumerLimits `json:"consumer_limits,omitempty"`
+
+	// Metadata is additional metadata for the Stream.
+	// Keys starting with `_nats` are reserved.
+	// NOTE: Metadata requires nats-server v2.10.0+
+	Metadata map[string]string `json:"metadata,omitempty"`
+}
+
+// SubjectTransformConfig is for applying a subject transform (to matching messages) before doing anything else when a new message is received.
+type SubjectTransformConfig struct {
+	Source      string `json:"src,omitempty"`
+	Destination string `json:"dest"`
 }
 
 // RePublish is for republishing messages once committed to a stream. The original
@@ -152,12 +171,13 @@ type Placement struct {
 
 // StreamSource dictates how streams can source from other streams.
 type StreamSource struct {
-	Name          string          `json:"name"`
-	OptStartSeq   uint64          `json:"opt_start_seq,omitempty"`
-	OptStartTime  *time.Time      `json:"opt_start_time,omitempty"`
-	FilterSubject string          `json:"filter_subject,omitempty"`
-	External      *ExternalStream `json:"external,omitempty"`
-	Domain        string          `json:"-"`
+	Name              string                   `json:"name"`
+	OptStartSeq       uint64                   `json:"opt_start_seq,omitempty"`
+	OptStartTime      *time.Time               `json:"opt_start_time,omitempty"`
+	FilterSubject     string                   `json:"filter_subject,omitempty"`
+	SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
+	External          *ExternalStream          `json:"external,omitempty"`
+	Domain            string                   `json:"-"`
 }
 
 // ExternalStream allows you to qualify access to a stream source in another
@@ -167,6 +187,13 @@ type ExternalStream struct {
 	DeliverPrefix string `json:"deliver,omitempty"`
 }
 
+// StreamConsumerLimits are the limits for a consumer on a stream.
+// These can be overridden on a per consumer basis.
+type StreamConsumerLimits struct {
+	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
+	MaxAckPending     int           `json:"max_ack_pending,omitempty"`
+}
+
 // Helper for copying when we do not want to change user's version.
 func (ss *StreamSource) copy() *StreamSource {
 	nss := *ss
@@ -407,6 +434,11 @@ func (js *js) upsertConsumer(stream, consumerName string, cfg *ConsumerConfig, o
 		}
 		return nil, info.Error
 	}
+
+	// check whether multiple filter subjects (if used) are reflected in the returned ConsumerInfo
+	if len(cfg.FilterSubjects) != 0 && len(info.Config.FilterSubjects) == 0 {
+		return nil, ErrConsumerMultipleFilterSubjectsNotSupported
+	}
 	return info.ConsumerInfo, nil
 }
 
@@ -780,6 +812,21 @@ func (js *js) AddStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error) {
 		return nil, resp.Error
 	}
 
+	// check that input subject transform (if used) is reflected in the returned ConsumerInfo
+	if cfg.SubjectTransform != nil && resp.StreamInfo.Config.SubjectTransform == nil {
+		return nil, ErrStreamSubjectTransformNotSupported
+	}
+	if len(cfg.Sources) != 0 {
+		if len(cfg.Sources) != len(resp.Config.Sources) {
+			return nil, ErrStreamSourceNotSupported
+		}
+		for i := range cfg.Sources {
+			if len(cfg.Sources[i].SubjectTransforms) != 0 && len(resp.Sources[i].SubjectTransforms) == 0 {
+				return nil, ErrStreamSourceMultipleSubjectTransformsNotSupported
+			}
+		}
+	}
+
 	return resp.StreamInfo, nil
 }
 
@@ -897,11 +944,13 @@ type StreamAlternate struct {
 
 // StreamSourceInfo shows information about an upstream stream source.
 type StreamSourceInfo struct {
-	Name     string          `json:"name"`
-	Lag      uint64          `json:"lag"`
-	Active   time.Duration   `json:"active"`
-	External *ExternalStream `json:"external"`
-	Error    *APIError       `json:"error"`
+	Name              string                   `json:"name"`
+	Lag               uint64                   `json:"lag"`
+	Active            time.Duration            `json:"active"`
+	External          *ExternalStream          `json:"external"`
+	Error             *APIError                `json:"error"`
+	FilterSubject     string                   `json:"filter_subject,omitempty"`
+	SubjectTransforms []SubjectTransformConfig `json:"subject_transforms,omitempty"`
 }
 
 // StreamState is information about the given stream.
@@ -973,6 +1022,23 @@ func (js *js) UpdateStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error
 		}
 		return nil, resp.Error
 	}
+
+	// check that input subject transform (if used) is reflected in the returned StreamInfo
+	if cfg.SubjectTransform != nil && resp.StreamInfo.Config.SubjectTransform == nil {
+		return nil, ErrStreamSubjectTransformNotSupported
+	}
+
+	if len(cfg.Sources) != 0 {
+		if len(cfg.Sources) != len(resp.Config.Sources) {
+			return nil, ErrStreamSourceNotSupported
+		}
+		for i := range cfg.Sources {
+			if len(cfg.Sources[i].SubjectTransforms) != 0 && len(resp.Sources[i].SubjectTransforms) == 0 {
+				return nil, ErrStreamSourceMultipleSubjectTransformsNotSupported
+			}
+		}
+	}
+
 	return resp.StreamInfo, nil
 }
 
diff --git a/kv.go b/kv.go
index 17c9e2f..7382f4d 100644
--- a/kv.go
+++ b/kv.go
@@ -432,14 +432,21 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 		scfg.Mirror = m
 		scfg.MirrorDirect = true
 	} else if len(cfg.Sources) > 0 {
-		// For now we do not allow direct subjects for sources. If that is desired a user could use stream API directly.
 		for _, ss := range cfg.Sources {
-			if !strings.HasPrefix(ss.Name, kvBucketNamePre) {
-				ss = ss.copy()
+			var sourceBucketName string
+			if strings.HasPrefix(ss.Name, kvBucketNamePre) {
+				sourceBucketName = ss.Name[len(kvBucketNamePre):]
+			} else {
+				sourceBucketName = ss.Name
 				ss.Name = fmt.Sprintf(kvBucketNameTmpl, ss.Name)
 			}
+
+			if ss.External == nil || sourceBucketName != cfg.Bucket {
+				ss.SubjectTransforms = []SubjectTransformConfig{{Source: fmt.Sprintf(kvSubjectsTmpl, sourceBucketName), Destination: fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}}
+			}
 			scfg.Sources = append(scfg.Sources, ss)
 		}
+		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
 	} else {
 		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
 	}
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 cee4ef3..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
 }
 
@@ -474,6 +510,10 @@ func (s *service) wrapConnectionEventCallbacks() {
 	s.natsHandlers.asyncErr = s.nc.ErrorHandler()
 	if s.natsHandlers.asyncErr != nil {
 		s.nc.SetErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
+			if sub == nil {
+				s.natsHandlers.asyncErr(c, sub, err)
+				return
+			}
 			endpoint, match := s.matchSubscriptionSubject(sub.Subject)
 			if !match {
 				s.natsHandlers.asyncErr(c, sub, err)
@@ -496,6 +536,9 @@ func (s *service) wrapConnectionEventCallbacks() {
 		})
 	} else {
 		s.nc.SetErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
+			if sub == nil {
+				return
+			}
 			endpoint, match := s.matchSubscriptionSubject(sub.Subject)
 			if !match {
 				return
@@ -651,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,
 		})
 	}
 
@@ -680,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,
@@ -731,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 queueGroupName(customQG, parentQG string) string {
+	queueGroup := customQG
+	if queueGroup == "" {
+		if parentQG != "" {
+			queueGroup = parentQG
+		} else {
+			queueGroup = DefaultQueueGroup
+		}
+	}
+	return queueGroup
 }
 
-func (g *group) AddGroup(name string) Group {
+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)
@@ -745,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,
 	}
 }
 
@@ -810,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 fca93f6..82b7973 100644
--- a/nats.go
+++ b/nats.go
@@ -47,7 +47,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.28.0"
+	Version                   = "1.30.2"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -61,6 +61,7 @@ const (
 	DefaultReconnectBufSize   = 8 * 1024 * 1024 // 8MB
 	RequestChanLen            = 8
 	DefaultDrainTimeout       = 30 * time.Second
+	DefaultFlusherTimeout     = time.Minute
 	LangString                = "go"
 )
 
@@ -154,6 +155,7 @@ func GetDefaultOptions() Options {
 		SubChanLen:         DefaultMaxChanLen,
 		ReconnectBufSize:   DefaultReconnectBufSize,
 		DrainTimeout:       DefaultDrainTimeout,
+		FlusherTimeout:     DefaultFlusherTimeout,
 	}
 }
 
@@ -356,6 +358,7 @@ type Options struct {
 
 	// FlusherTimeout is the maximum time to wait for write operations
 	// to the underlying connection to complete (including the flusher loop).
+	// Defaults to 1m.
 	FlusherTimeout time.Duration
 
 	// PingInterval is the period at which the client will be sending ping
@@ -2501,6 +2504,9 @@ func (nc *Conn) sendConnect() error {
 	// Construct the CONNECT protocol string
 	cProto, err := nc.connectProto()
 	if err != nil {
+		if !nc.initc && nc.Opts.AsyncErrorCB != nil {
+			nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
+		}
 		return err
 	}
 
@@ -3162,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 {
diff --git a/object.go b/object.go
index b025072..f6ba8fb 100644
--- a/object.go
+++ b/object.go
@@ -34,10 +34,6 @@ import (
 )
 
 // ObjectStoreManager creates, loads and deletes Object Stores
-//
-// Notice: Experimental Preview
-//
-// This functionality is EXPERIMENTAL and may be changed in later releases.
 type ObjectStoreManager interface {
 	// ObjectStore will look up and bind to an existing object store instance.
 	ObjectStore(bucket string) (ObjectStore, error)
@@ -53,10 +49,6 @@ type ObjectStoreManager interface {
 
 // ObjectStore is a blob store capable of storing large objects efficiently in
 // JetStream streams
-//
-// Notice: Experimental Preview
-//
-// This functionality is EXPERIMENTAL and may be changed in later releases.
 type ObjectStore interface {
 	// Put will place the contents from the reader into a new object.
 	Put(obj *ObjectMeta, reader io.Reader, opts ...ObjectOpt) (*ObjectInfo, error)
@@ -157,6 +149,10 @@ type ObjectStoreConfig struct {
 	Storage     StorageType   `json:"storage,omitempty"`
 	Replicas    int           `json:"num_replicas,omitempty"`
 	Placement   *Placement    `json:"placement,omitempty"`
+
+	// Bucket-specific metadata
+	// NOTE: Metadata requires nats-server v2.10.0+
+	Metadata map[string]string `json:"metadata,omitempty"`
 }
 
 type ObjectStoreStatus interface {
@@ -176,6 +172,8 @@ type ObjectStoreStatus interface {
 	Size() uint64
 	// BackingStore provides details about the underlying storage
 	BackingStore() string
+	// Metadata is the user supplied metadata for the bucket
+	Metadata() map[string]string
 }
 
 // ObjectMetaOptions
@@ -186,9 +184,10 @@ type ObjectMetaOptions struct {
 
 // ObjectMeta is high level information about an object.
 type ObjectMeta struct {
-	Name        string `json:"name"`
-	Description string `json:"description,omitempty"`
-	Headers     Header `json:"headers,omitempty"`
+	Name        string            `json:"name"`
+	Description string            `json:"description,omitempty"`
+	Headers     Header            `json:"headers,omitempty"`
+	Metadata    map[string]string `json:"metadata,omitempty"`
 
 	// Optional options.
 	Opts *ObjectMetaOptions `json:"options,omitempty"`
@@ -280,6 +279,7 @@ func (js *js) CreateObjectStore(cfg *ObjectStoreConfig) (ObjectStore, error) {
 		Discard:     DiscardNew,
 		AllowRollup: true,
 		AllowDirect: true,
+		Metadata:    cfg.Metadata,
 	}
 
 	// Create our stream.
@@ -974,6 +974,7 @@ func (obs *obs) UpdateMeta(name string, meta *ObjectMeta) error {
 	info.Name = meta.Name
 	info.Description = meta.Description
 	info.Headers = meta.Headers
+	info.Metadata = meta.Metadata
 
 	// Prepare the meta message
 	if err = publishMeta(info, obs.js); err != nil {
@@ -1197,6 +1198,9 @@ func (s *ObjectBucketStatus) Size() uint64 { return s.nfo.State.Bytes }
 // BackingStore indicates what technology is used for storage of the bucket
 func (s *ObjectBucketStatus) BackingStore() string { return "JetStream" }
 
+// Metadata is the metadata supplied when creating the bucket
+func (s *ObjectBucketStatus) Metadata() map[string]string { return s.nfo.Config.Metadata }
+
 // StreamInfo is the stream info retrieved to create the status
 func (s *ObjectBucketStatus) StreamInfo() *StreamInfo { return s.nfo }
 


Test patch:
diff --git a/example_test.go b/example_test.go
index d54c8e2..11ebe8f 100644
--- a/example_test.go
+++ b/example_test.go
@@ -675,6 +675,11 @@ func ExampleSubOpt() {
 	js.Subscribe("foo", func(msg *nats.Msg) {
 		fmt.Printf("Received a message: %s\n", string(msg.Data))
 	}, nats.Durable("FOO"), nats.SkipConsumerLookup())
+
+	// Use multiple subject filters.
+	js.Subscribe("", func(msg *nats.Msg) {
+		fmt.Printf("Received a message: %s\n", string(msg.Data))
+	}, nats.Durable("FOO"), nats.ConsumerFilterSubjects("foo", "bar"), nats.BindStream("test_stream"))
 }
 
 func ExampleMaxWait() {
diff --git a/go_test.mod b/go_test.mod
index af69c5a..8902c1e 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -4,19 +4,19 @@ go 1.19
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/klauspost/compress v1.16.5
-	github.com/nats-io/nats-server/v2 v2.9.19
-	github.com/nats-io/nkeys v0.4.4
+	github.com/klauspost/compress v1.17.0
+	github.com/nats-io/nats-server/v2 v2.10.0
+	github.com/nats-io/nkeys v0.4.5
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/goleak v1.2.1
-	golang.org/x/text v0.9.0
+	golang.org/x/text v0.13.0
 	google.golang.org/protobuf v1.23.0
 )
 
 require (
 	github.com/minio/highwayhash v1.0.2 // indirect
-	github.com/nats-io/jwt/v2 v2.4.1 // indirect
-	golang.org/x/crypto v0.9.0 // indirect
-	golang.org/x/sys v0.8.0 // indirect
+	github.com/nats-io/jwt/v2 v2.5.2 // indirect
+	golang.org/x/crypto v0.13.0 // indirect
+	golang.org/x/sys v0.12.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index b9c9fd4..ce4ba92 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -10,29 +10,30 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
-github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
+github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
-github.com/nats-io/jwt/v2 v2.4.1 h1:Y35W1dgbbz2SQUYDPCaclXcuqleVmpbRa7646Jf2EX4=
-github.com/nats-io/jwt/v2 v2.4.1/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
-github.com/nats-io/nats-server/v2 v2.9.19 h1:OF9jSKZGo425C/FcVVIvNgpd36CUe7aVTTXEZRJk6kA=
-github.com/nats-io/nats-server/v2 v2.9.19/go.mod h1:aTb/xtLCGKhfTFLxP591CMWfkdgBmcUUSkiSOe5A3gw=
-github.com/nats-io/nkeys v0.4.4 h1:xvBJ8d69TznjcQl9t6//Q5xXuVhyYiSos6RPtvQNTwA=
-github.com/nats-io/nkeys v0.4.4/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
+github.com/nats-io/jwt/v2 v2.5.2 h1:DhGH+nKt+wIkDxM6qnVSKjokq5t59AZV5HRcFW0zJwU=
+github.com/nats-io/jwt/v2 v2.5.2/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
+github.com/nats-io/nats-server/v2 v2.10.0 h1:rcU++Hzo+wARxtJugrV3J5z5iGdHeVG8tT8Chb3bKDg=
+github.com/nats-io/nats-server/v2 v2.10.0/go.mod h1:3PMvMSu2cuK0J9YInRLWdFpFsswKKGUS77zVSAudRto=
+github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk=
+github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
 go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
-golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
-golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
+golang.org/x/crypto v0.13.0 h1:mvySKfSWJ+UKUii46M40LOvyWfN0s2U+46/jDd0e6Ck=
+golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
+golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
-golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
-golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/sys v0.12.0 h1:CM0HF96J0hcLAwsHPJZjfdNzs0gftsLfgKt57wWHJ0o=
+golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
 golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
diff --git a/jetstream/test/jetstream_test.go b/jetstream/test/jetstream_test.go
index c41312c..410c7f7 100644
--- a/jetstream/test/jetstream_test.go
+++ b/jetstream/test/jetstream_test.go
@@ -205,6 +205,7 @@ func TestCreateStream(t *testing.T) {
 		name      string
 		stream    string
 		subject   string
+		metadata  map[string]string
 		timeout   time.Duration
 		withError error
 	}{
@@ -214,10 +215,30 @@ func TestCreateStream(t *testing.T) {
 			timeout: 10 * time.Second,
 			subject: "FOO.123",
 		},
+		{
+			name:   "create stream with metadata",
+			stream: "foo_meta",
+			metadata: map[string]string{
+				"foo":  "bar",
+				"name": "test",
+			},
+			timeout: 10 * time.Second,
+			subject: "FOO.meta",
+		},
+		{
+			name:   "create stream with metadata, reserved prefix",
+			stream: "foo_meta1",
+			metadata: map[string]string{
+				"foo":           "bar",
+				"_nats_version": "2.10.0",
+			},
+			timeout: 10 * time.Second,
+			subject: "FOO.meta1",
+		},
 		{
 			name:    "with empty context",
-			stream:  "foo",
-			subject: "FOO.123",
+			stream:  "foo_empty_ctx",
+			subject: "FOO.ctx",
 		},
 		{
 			name:      "invalid stream name",
@@ -270,7 +291,7 @@ func TestCreateStream(t *testing.T) {
 				ctx, cancel = context.WithTimeout(context.Background(), test.timeout)
 				defer cancel()
 			}
-			_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: test.stream, Subjects: []string{test.subject}})
+			s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: test.stream, Subjects: []string{test.subject}, Metadata: test.metadata})
 			if test.withError != nil {
 				if !errors.Is(err, test.withError) {
 					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
@@ -280,6 +301,9 @@ func TestCreateStream(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
+			if !reflect.DeepEqual(s.CachedInfo().Config.Metadata, test.metadata) {
+				t.Fatalf("Invalid metadata; want: %v, got: %v", test.metadata, s.CachedInfo().Config.Metadata)
+			}
 		})
 	}
 }
@@ -431,11 +455,142 @@ func TestCreateStreamMirrorCrossDomains(t *testing.T) {
 	}
 }
 
+func TestCreateOrUpdateStream(t *testing.T) {
+	tests := []struct {
+		name          string
+		stream        string
+		subject       string
+		timeout       time.Duration
+		withError     error
+		withInfoCheck bool
+	}{
+		{
+			name:          "create stream ok",
+			stream:        "foo",
+			timeout:       10 * time.Second,
+			subject:       "FOO.1",
+			withInfoCheck: false,
+		},
+		{
+			name:          "create stream empty context",
+			stream:        "foo-o",
+			subject:       "FOO.12",
+			withInfoCheck: false,
+		},
+		{
+			name:          "create stream invalid stream name",
+			stream:        "foo.123",
+			subject:       "FOO-123",
+			timeout:       10 * time.Second,
+			withError:     jetstream.ErrInvalidStreamName,
+			withInfoCheck: false,
+		},
+		{
+			name:          "create stream stream name required",
+			stream:        "",
+			subject:       "FOO-1234",
+			timeout:       10 * time.Second,
+			withError:     jetstream.ErrStreamNameRequired,
+			withInfoCheck: false,
+		},
+		{
+			name:          "update stream ok",
+			stream:        "foo",
+			subject:       "BAR-123",
+			timeout:       10 * time.Second,
+			withInfoCheck: true,
+		},
+		{
+			name:          "create stream context timeout",
+			stream:        "foo",
+			subject:       "BAR-1234",
+			timeout:       1 * time.Microsecond,
+			withError:     context.DeadlineExceeded,
+			withInfoCheck: false,
+		},
+		{
+			name:          "update stream with empty context",
+			stream:        "sample-foo-1",
+			subject:       "SAMPLE-FOO-123",
+			withInfoCheck: true,
+		},
+		{
+			name:          "update stream invalid stream name",
+			stream:        "sample-foo.123",
+			subject:       "SAMPLE-FOO-1234",
+			timeout:       10 * time.Second,
+			withError:     jetstream.ErrInvalidStreamName,
+			withInfoCheck: true,
+		},
+		{
+			name:          "update stream stream name required",
+			stream:        "",
+			subject:       "SAMPLE-FOO-123",
+			timeout:       10 * time.Second,
+			withError:     jetstream.ErrStreamNameRequired,
+			withInfoCheck: true,
+		},
+		{
+			name:          "update stream context timeout",
+			stream:        "sample-foo-2",
+			subject:       "SAMPLE-FOO-123456",
+			timeout:       1 * time.Microsecond,
+			withError:     context.DeadlineExceeded,
+			withInfoCheck: true,
+		},
+	}
+
+	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()
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ctx := context.Background()
+			if test.timeout > 0 {
+				var cancel context.CancelFunc
+				ctx, cancel = context.WithTimeout(context.Background(), test.timeout)
+				defer cancel()
+			}
+			s, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: test.stream, Subjects: []string{test.subject}})
+			if test.withError != nil {
+				if !errors.Is(err, test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			if test.withInfoCheck {
+				info, err := s.Info(ctx)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				if len(info.Config.Subjects) != 1 || info.Config.Subjects[0] != test.subject {
+					t.Fatalf("Invalid stream subjects after update: %v", info.Config.Subjects)
+				}
+			}
+		})
+	}
+}
+
 func TestUpdateStream(t *testing.T) {
 	tests := []struct {
 		name      string
 		stream    string
 		subject   string
+		metadata  map[string]string
 		timeout   time.Duration
 		withError error
 	}{
@@ -450,6 +605,16 @@ func TestUpdateStream(t *testing.T) {
 			stream:  "foo",
 			subject: "FOO.123",
 		},
+		{
+			name:    "update existing, add metadata",
+			stream:  "foo",
+			subject: "BAR.123",
+			metadata: map[string]string{
+				"foo":  "bar",
+				"name": "test",
+			},
+			timeout: 10 * time.Second,
+		},
 		{
 			name:      "invalid stream name",
 			stream:    "foo.123",
@@ -504,7 +669,7 @@ func TestUpdateStream(t *testing.T) {
 				ctx, cancel = context.WithTimeout(context.Background(), test.timeout)
 				defer cancel()
 			}
-			s, err := js.UpdateStream(ctx, jetstream.StreamConfig{Name: test.stream, Subjects: []string{test.subject}})
+			s, err := js.UpdateStream(ctx, jetstream.StreamConfig{Name: test.stream, Subjects: []string{test.subject}, Metadata: test.metadata})
 			if test.withError != nil {
 				if !errors.Is(err, test.withError) {
 					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
@@ -521,6 +686,9 @@ func TestUpdateStream(t *testing.T) {
 			if len(info.Config.Subjects) != 1 || info.Config.Subjects[0] != test.subject {
 				t.Fatalf("Invalid stream subjects after update: %v", info.Config.Subjects)
 			}
+			if !reflect.DeepEqual(info.Config.Metadata, test.metadata) {
+				t.Fatalf("Invalid metadata; want: %v, got: %v", test.metadata, info.Config.Metadata)
+			}
 		})
 	}
 }
@@ -1090,6 +1258,187 @@ func TestJetStream_CreateOrUpdateConsumer(t *testing.T) {
 	}
 }
 
+func TestJetStream_CreateConsumer(t *testing.T) {
+	tests := []struct {
+		name           string
+		consumerConfig jetstream.ConsumerConfig
+		shouldCreate   bool
+		stream         string
+		withError      error
+	}{
+		{
+			name:           "create consumer",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
+			stream:         "foo",
+			shouldCreate:   true,
+		},
+		{
+			name:           "consumer already exists, error",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", Description: "test consumer"},
+			stream:         "foo",
+			withError:      jetstream.ErrConsumerExists,
+		},
+		{
+			name:           "stream does not exist",
+			stream:         "abc",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
+			withError:      jetstream.ErrStreamNotFound,
+		},
+	}
+
+	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)
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var sub *nats.Subscription
+			if test.consumerConfig.FilterSubject != "" {
+				sub, err = nc.SubscribeSync(fmt.Sprintf("$JS.API.CONSUMER.CREATE.foo.*.%s", test.consumerConfig.FilterSubject))
+			} else {
+				sub, err = nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo.*")
+			}
+			c, err := js.CreateConsumer(ctx, test.stream, test.consumerConfig)
+			if test.withError != nil {
+				if !errors.Is(err, test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if test.shouldCreate {
+				if _, err := sub.NextMsgWithContext(ctx); err != nil {
+					t.Fatalf("Expected request on %s; got %s", sub.Subject, err)
+				}
+			}
+			ci, err := s.Consumer(ctx, c.CachedInfo().Name)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if ci.CachedInfo().Config.AckPolicy != test.consumerConfig.AckPolicy {
+				t.Fatalf("Invalid ack policy; want: %s; got: %s", test.consumerConfig.AckPolicy, ci.CachedInfo().Config.AckPolicy)
+			}
+			if !reflect.DeepEqual(test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects) {
+				t.Fatalf("Invalid filter subjects; want: %v; got: %v", test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects)
+			}
+		})
+	}
+}
+
+func TestJetStream_UpdateConsumer(t *testing.T) {
+	tests := []struct {
+		name           string
+		consumerConfig jetstream.ConsumerConfig
+		shouldUpdate   bool
+		stream         string
+		withError      error
+	}{
+		{
+			name:           "update consumer",
+			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", Description: "updated consumer"},
+			stream:         "foo",
+			shouldUpdate:   true,
+		},
+		{
+			name:           "illegal update",
+			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", AckPolicy: jetstream.AckNonePolicy},
+			stream:         "foo",
+			withError:      jetstream.ErrConsumerCreate,
+		},
+		{
+			name:           "consumer does not exist",
+			consumerConfig: jetstream.ConsumerConfig{Name: "abc"},
+			stream:         "foo",
+			withError:      jetstream.ErrConsumerDoesNotExist,
+		},
+		{
+			name:           "stream does not exist",
+			consumerConfig: jetstream.ConsumerConfig{Name: "testcons"},
+			stream:         "abc",
+			withError:      jetstream.ErrStreamNotFound,
+		},
+	}
+
+	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)
+	}
+
+	_, err = s.CreateConsumer(ctx, jetstream.ConsumerConfig{Name: "testcons"})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var sub *nats.Subscription
+			if test.consumerConfig.FilterSubject != "" {
+				sub, err = nc.SubscribeSync(fmt.Sprintf("$JS.API.CONSUMER.CREATE.foo.*.%s", test.consumerConfig.FilterSubject))
+			} else {
+				sub, err = nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo.*")
+			}
+			c, err := js.UpdateConsumer(ctx, test.stream, test.consumerConfig)
+			if test.withError != nil {
+				if !errors.Is(err, test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if test.shouldUpdate {
+				if _, err := sub.NextMsgWithContext(ctx); err != nil {
+					t.Fatalf("Expected request on %s; got %s", sub.Subject, err)
+				}
+			}
+			ci, err := s.Consumer(ctx, c.CachedInfo().Name)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if ci.CachedInfo().Config.AckPolicy != test.consumerConfig.AckPolicy {
+				t.Fatalf("Invalid ack policy; want: %s; got: %s", test.consumerConfig.AckPolicy, ci.CachedInfo().Config.AckPolicy)
+			}
+			if !reflect.DeepEqual(test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects) {
+				t.Fatalf("Invalid filter subjects; want: %v; got: %v", test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects)
+			}
+		})
+	}
+}
+
 func TestJetStream_Consumer(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -1249,12 +1598,12 @@ func TestJetStream_DeleteConsumer(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+	defer nc.Close()
 
 	js, err := jetstream.New(nc)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	defer nc.Close()
 
 	s, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
 	if err != nil {
@@ -1388,3 +1737,223 @@ func TestStreamNameBySubject(t *testing.T) {
 		})
 	}
 }
+
+func TestJetStreamTransform(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	ctx := context.Background()
+	_, err := js.CreateStream(ctx, jetstream.StreamConfig{
+		Name:             "ORIGIN",
+		Subjects:         []string{"test"},
+		SubjectTransform: &jetstream.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+		Storage:          jetstream.MemoryStorage,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	err = nc.Publish("test", []byte("1"))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	sourcingStream, err := js.CreateStream(ctx, jetstream.StreamConfig{
+		Subjects: []string{},
+		Name:     "SOURCING",
+		Sources:  []*jetstream.StreamSource{{Name: "ORIGIN", SubjectTransforms: []jetstream.SubjectTransformConfig{{Source: ">", Destination: "fromtest.>"}}}},
+		Storage:  jetstream.MemoryStorage,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	cons, err := sourcingStream.CreateConsumer(ctx, jetstream.ConsumerConfig{FilterSubject: "fromtest.>", MemoryStorage: true})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	m, err := cons.Next()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if m.Subject() != "fromtest.transformed.test" {
+		t.Fatalf("the subject of the message doesn't match the expected fromtest.transformed.test: %s", m.Subject())
+	}
+}
+
+func TestStreamConfigMatches(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	cfg := jetstream.StreamConfig{
+		Name:                 "stream",
+		Description:          "desc",
+		Subjects:             []string{"foo.*"},
+		Retention:            jetstream.WorkQueuePolicy,
+		MaxConsumers:         10,
+		MaxMsgs:              100,
+		MaxBytes:             1000,
+		Discard:              jetstream.DiscardNew,
+		DiscardNewPerSubject: true,
+		MaxAge:               100 * time.Second,
+		MaxMsgsPerSubject:    1000,
+		MaxMsgSize:           10000,
+		Storage:              jetstream.MemoryStorage,
+		Replicas:             1,
+		NoAck:                true,
+		Duplicates:           10 * time.Second,
+		Sealed:               false,
+		DenyDelete:           true,
+		DenyPurge:            false,
+		AllowRollup:          true,
+		Compression:          jetstream.S2Compression,
+		FirstSeq:             5,
+		SubjectTransform:     &jetstream.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+		RePublish: &jetstream.RePublish{
+			Source:      ">",
+			Destination: "RP.>",
+			HeadersOnly: true,
+		},
+		AllowDirect: true,
+		ConsumerLimits: jetstream.StreamConsumerLimits{
+			InactiveThreshold: 10 * time.Second,
+			MaxAckPending:     500,
+		},
+		Metadata: map[string]string{
+			"foo": "bar",
+		},
+	}
+
+	s, err := js.CreateStream(context.Background(), cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(s.CachedInfo().Config, cfg) {
+		t.Fatalf("StreamConfig doesn't match: %#v", s.CachedInfo().Config)
+	}
+
+	cfgMirror := jetstream.StreamConfig{
+		Name:              "mirror",
+		MaxConsumers:      10,
+		MaxMsgs:           100,
+		MaxBytes:          1000,
+		MaxAge:            100 * time.Second,
+		MaxMsgsPerSubject: 1000,
+		MaxMsgSize:        10000,
+		Replicas:          1,
+		Duplicates:        10 * time.Second,
+		Mirror: &jetstream.StreamSource{
+			Name:        "stream",
+			OptStartSeq: 10,
+			SubjectTransforms: []jetstream.SubjectTransformConfig{
+				{Source: ">", Destination: "transformed.>"},
+			},
+		},
+		MirrorDirect:     true,
+		SubjectTransform: &jetstream.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+	}
+
+	s, err = js.CreateStream(context.Background(), cfgMirror)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(s.CachedInfo().Config, cfgMirror) {
+		t.Fatalf("StreamConfig doesn't match: %#v", s.CachedInfo().Config)
+	}
+
+	cfgSourcing := jetstream.StreamConfig{
+		Name:              "sourcing",
+		Subjects:          []string{"BAR"},
+		MaxConsumers:      10,
+		MaxMsgs:           100,
+		MaxBytes:          1000,
+		MaxAge:            100 * time.Second,
+		MaxMsgsPerSubject: 1000,
+		MaxMsgSize:        10000,
+		Replicas:          1,
+		Duplicates:        10 * time.Second,
+		Sources: []*jetstream.StreamSource{
+			{
+				Name:        "stream",
+				OptStartSeq: 10,
+				SubjectTransforms: []jetstream.SubjectTransformConfig{
+					{Source: ">", Destination: "transformed.>"},
+				},
+			},
+		},
+		SubjectTransform: &jetstream.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+	}
+
+	s, err = js.CreateStream(context.Background(), cfgSourcing)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(s.CachedInfo().Config, cfgSourcing) {
+		t.Fatalf("StreamConfig doesn't match: %#v", s.CachedInfo().Config)
+	}
+}
+
+func TestConsumerConfigMatches(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	s, err := js.CreateStream(context.Background(), jetstream.StreamConfig{
+		Name:     "FOO",
+		Subjects: []string{"foo.*"},
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	cfg := jetstream.ConsumerConfig{
+		Name:               "cons",
+		Durable:            "cons",
+		Description:        "test",
+		DeliverPolicy:      jetstream.DeliverByStartSequencePolicy,
+		OptStartSeq:        5,
+		AckPolicy:          jetstream.AckAllPolicy,
+		AckWait:            1 * time.Second,
+		MaxDeliver:         5,
+		BackOff:            []time.Duration{1 * time.Second, 2 * time.Second, 3 * time.Second},
+		ReplayPolicy:       jetstream.ReplayOriginalPolicy,
+		SampleFrequency:    "50%",
+		MaxWaiting:         100,
+		MaxAckPending:      1000,
+		HeadersOnly:        true,
+		MaxRequestBatch:    100,
+		MaxRequestExpires:  10 * time.Second,
+		MaxRequestMaxBytes: 1000,
+		InactiveThreshold:  20 * time.Second,
+		Replicas:           1,
+		MemoryStorage:      true,
+		FilterSubjects:     []string{"foo.1", "foo.2"},
+		Metadata: map[string]string{
+			"foo": "bar",
+		},
+	}
+
+	c, err := s.CreateConsumer(context.Background(), cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(c.CachedInfo().Config, cfg) {
+		t.Fatalf("ConsumerConfig doesn't match")
+	}
+}
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/publish_test.go b/jetstream/test/publish_test.go
index 80ddc71..ac7a34c 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1233,6 +1233,43 @@ func TestPublishMsgAsyncWithPendingMsgs(t *testing.T) {
 			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrTooManyStalledMsgs, err)
 		}
 	})
+
+	t.Run("with server restart", 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()
+
+		_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		go func() {
+			for i := 0; i < 50; i++ {
+				_, _ = js.PublishAsync("FOO.1", []byte("msg"))
+			}
+		}()
+		srv = restartBasicJSServer(t, srv)
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(10 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+	})
 }
 
 func TestPublishAsyncResetPendingOnReconnect(t *testing.T) {
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/jetstream/test/stream_test.go b/jetstream/test/stream_test.go
index b90a79b..bb85f67 100644
--- a/jetstream/test/stream_test.go
+++ b/jetstream/test/stream_test.go
@@ -53,6 +53,21 @@ func TestCreateOrUpdateConsumer(t *testing.T) {
 			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", "FOO.B"}},
 			shouldCreate:   true,
 		},
+		{
+			name:           "with multiple filter subjects, overlapping subjects",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.*", "FOO.B"}},
+			withError:      jetstream.ErrOverlappingFilterSubjects,
+		},
+		{
+			name:           "with multiple filter subjects and filter subject provided",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", "FOO.B"}, FilterSubject: "FOO.C"},
+			withError:      jetstream.ErrDuplicateFilterSubjects,
+		},
+		{
+			name:           "with empty subject in FilterSubjects",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", ""}},
+			withError:      jetstream.ErrEmptyFilter,
+		},
 		{
 			name:           "consumer already exists, update",
 			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", Description: "test consumer"},
@@ -119,6 +134,224 @@ func TestCreateOrUpdateConsumer(t *testing.T) {
 			if ci.CachedInfo().Config.AckPolicy != test.consumerConfig.AckPolicy {
 				t.Fatalf("Invalid ack policy; want: %s; got: %s", test.consumerConfig.AckPolicy, ci.CachedInfo().Config.AckPolicy)
 			}
+			if !reflect.DeepEqual(test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects) {
+				t.Fatalf("Invalid filter subjects; want: %v; got: %v", test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects)
+			}
+		})
+	}
+}
+
+func TestCreateConsumer(t *testing.T) {
+	tests := []struct {
+		name           string
+		consumerConfig jetstream.ConsumerConfig
+		shouldCreate   bool
+		withError      error
+	}{
+		{
+			name:           "create durable pull consumer",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
+			shouldCreate:   true,
+		},
+		{
+			name:           "idempotent create, no error",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur"},
+			shouldCreate:   true,
+		},
+		{
+			name:           "create ephemeral pull consumer",
+			consumerConfig: jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy},
+			shouldCreate:   true,
+		},
+		{
+			name:           "with filter subject",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubject: "FOO.A"},
+			shouldCreate:   true,
+		},
+		{
+			name:           "with metadata",
+			consumerConfig: jetstream.ConsumerConfig{Metadata: map[string]string{"foo": "bar", "baz": "quux"}},
+			shouldCreate:   true,
+		},
+		{
+			name:           "with multiple filter subjects",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", "FOO.B"}},
+			shouldCreate:   true,
+		},
+		{
+			name:           "with multiple filter subjects, overlapping subjects",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.*", "FOO.B"}},
+			withError:      jetstream.ErrOverlappingFilterSubjects,
+		},
+		{
+			name:           "with multiple filter subjects and filter subject provided",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", "FOO.B"}, FilterSubject: "FOO.C"},
+			withError:      jetstream.ErrDuplicateFilterSubjects,
+		},
+		{
+			name:           "with empty subject in FilterSubjects",
+			consumerConfig: jetstream.ConsumerConfig{FilterSubjects: []string{"FOO.A", ""}},
+			withError:      jetstream.ErrEmptyFilter,
+		},
+		{
+			name:           "consumer already exists, error",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur", Description: "test consumer"},
+			withError:      jetstream.ErrConsumerExists,
+		},
+		{
+			name:           "invalid durable name",
+			consumerConfig: jetstream.ConsumerConfig{Durable: "dur.123"},
+			withError:      jetstream.ErrInvalidConsumerName,
+		},
+	}
+
+	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)
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var sub *nats.Subscription
+			if test.consumerConfig.FilterSubject != "" {
+				sub, err = nc.SubscribeSync(fmt.Sprintf("$JS.API.CONSUMER.CREATE.foo.*.%s", test.consumerConfig.FilterSubject))
+			} else {
+				sub, err = nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo.*")
+			}
+			c, err := s.CreateConsumer(ctx, test.consumerConfig)
+			if test.withError != nil {
+				if !errors.Is(err, test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if test.shouldCreate {
+				if _, err := sub.NextMsgWithContext(ctx); err != nil {
+					t.Fatalf("Expected request on %s; got %s", sub.Subject, err)
+				}
+			}
+			ci, err := s.Consumer(ctx, c.CachedInfo().Name)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if ci.CachedInfo().Config.AckPolicy != test.consumerConfig.AckPolicy {
+				t.Fatalf("Invalid ack policy; want: %s; got: %s", test.consumerConfig.AckPolicy, ci.CachedInfo().Config.AckPolicy)
+			}
+			if !reflect.DeepEqual(test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects) {
+				t.Fatalf("Invalid filter subjects; want: %v; got: %v", test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects)
+			}
+			if !reflect.DeepEqual(test.consumerConfig.Metadata, ci.CachedInfo().Config.Metadata) {
+				t.Fatalf("Invalid metadata; want: %v; got: %v", test.consumerConfig.Metadata, ci.CachedInfo().Config.Metadata)
+			}
+		})
+	}
+}
+
+func TestUpdateConsumer(t *testing.T) {
+	tests := []struct {
+		name           string
+		consumerConfig jetstream.ConsumerConfig
+		shouldUpdate   bool
+		withError      error
+	}{
+		{
+			name:           "update consumer",
+			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", Description: "updated consumer"},
+			shouldUpdate:   true,
+		},
+		{
+			name:           "update consumer, with metadata",
+			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", Description: "updated consumer", Metadata: map[string]string{"foo": "bar", "baz": "quux"}},
+			shouldUpdate:   true,
+		},
+		{
+			name:           "illegal update",
+			consumerConfig: jetstream.ConsumerConfig{Name: "testcons", AckPolicy: jetstream.AckNonePolicy},
+			withError:      jetstream.ErrConsumerCreate,
+		},
+		{
+			name:           "consumer does not exist",
+			consumerConfig: jetstream.ConsumerConfig{Name: "abc"},
+			withError:      jetstream.ErrConsumerDoesNotExist,
+		},
+	}
+
+	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)
+	}
+
+	_, err = s.CreateConsumer(ctx, jetstream.ConsumerConfig{Name: "testcons"})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var sub *nats.Subscription
+			if test.consumerConfig.FilterSubject != "" {
+				sub, err = nc.SubscribeSync(fmt.Sprintf("$JS.API.CONSUMER.CREATE.foo.*.%s", test.consumerConfig.FilterSubject))
+			} else {
+				sub, err = nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo.*")
+			}
+			c, err := s.UpdateConsumer(ctx, test.consumerConfig)
+			if test.withError != nil {
+				if !errors.Is(err, test.withError) {
+					t.Fatalf("Expected error: %v; got: %v", test.withError, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if test.shouldUpdate {
+				if _, err := sub.NextMsgWithContext(ctx); err != nil {
+					t.Fatalf("Expected request on %s; got %s", sub.Subject, err)
+				}
+			}
+			ci, err := s.Consumer(ctx, c.CachedInfo().Name)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if ci.CachedInfo().Config.AckPolicy != test.consumerConfig.AckPolicy {
+				t.Fatalf("Invalid ack policy; want: %s; got: %s", test.consumerConfig.AckPolicy, ci.CachedInfo().Config.AckPolicy)
+			}
+			if !reflect.DeepEqual(test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects) {
+				t.Fatalf("Invalid filter subjects; want: %v; got: %v", test.consumerConfig.FilterSubjects, ci.CachedInfo().Config.FilterSubjects)
+			}
 		})
 	}
 }
diff --git a/js_test.go b/js_test.go
index f01547e..a36d962 100644
--- a/js_test.go
+++ b/js_test.go
@@ -1276,3 +1276,51 @@ func TestStreamNameBySubject(t *testing.T) {
 		}
 	}
 }
+
+func TestJetStreamTransform(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&StreamConfig{
+		Name:             "ORIGIN",
+		Subjects:         []string{"test"},
+		SubjectTransform: &SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+		Storage:          MemoryStorage,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	err = nc.Publish("test", []byte("1"))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	_, err = js.AddStream(&StreamConfig{
+		Subjects: []string{},
+		Name:     "SOURCING",
+		Sources:  []*StreamSource{{Name: "ORIGIN", SubjectTransforms: []SubjectTransformConfig{{Source: ">", Destination: "fromtest.>"}}}},
+		Storage:  MemoryStorage,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Create a sync subscription with an in-memory ephemeral consumer.
+	sub, err := js.SubscribeSync("fromtest.>", ConsumerMemoryStorage(), BindStream("SOURCING"))
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+
+	m, err := sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if m.Subject != "fromtest.transformed.test" {
+		t.Fatalf("the subject of the message doesn't match the expected fromtest.transformed.test: %s", m.Subject)
+	}
+
+}
diff --git a/kv_test.go b/kv_test.go
index 0eb96cb..3e0d484 100644
--- a/kv_test.go
+++ b/kv_test.go
@@ -251,3 +251,66 @@ func TestKeyValueCreate(t *testing.T) {
 		t.Fatalf("Unexpected error code, got: %v", kerr.APIError().ErrorCode)
 	}
 }
+
+func TestKeyValueSourcing(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	kvA, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "A"})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+
+	_, err = kvA.Create("keyA", []byte("1"))
+	if err != nil {
+		t.Fatalf("Error creating key: %v", err)
+	}
+
+	if _, err := kvA.Get("keyA"); err != nil {
+		t.Fatalf("Got error getting keyA from A: %v", err)
+	}
+
+	kvB, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "B"})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+
+	_, err = kvB.Create("keyB", []byte("1"))
+	if err != nil {
+		t.Fatalf("Error creating key: %v", err)
+	}
+
+	kvC, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "C", Sources: []*StreamSource{{Name: "A"}, {Name: "B"}}})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+
+	// Wait half a second to make sure it has time to populate the stream from it's sources
+	i := 0
+	for {
+		status, err := kvC.Status()
+		if err != nil {
+			t.Fatalf("Error getting bucket status: %v", err)
+		}
+		if status.Values() == 2 {
+			break
+		} else {
+			i++
+			if i > 3 {
+				t.Fatalf("Error sourcing bucket does not contain the expected number of values")
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	if _, err := kvC.Get("keyA"); err != nil {
+		t.Fatalf("Got error getting keyA from C: %v", err)
+	}
+
+	if _, err := kvC.Get("keyB"); err != nil {
+		t.Fatalf("Got error getting keyB from C: %v", err)
+	}
+}
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/nats_test.go b/nats_test.go
index 69cfb28..b70a190 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -1575,6 +1575,38 @@ func TestUserCredentialsChainedFile(t *testing.T) {
 	}
 }
 
+func TestReconnectMissingCredentials(t *testing.T) {
+	ts := runTrustServer()
+	defer ts.Shutdown()
+
+	chainedFile := createTmpFile(t, []byte(chained))
+	defer os.Remove(chainedFile)
+
+	url := fmt.Sprintf("nats://127.0.0.1:%d", TEST_PORT)
+	errs := make(chan error, 1)
+	nc, err := Connect(url, UserCredentials(chainedFile), ErrorHandler(func(_ *Conn, _ *Subscription, err error) {
+		errs <- err
+	}))
+	if err != nil {
+		t.Fatalf("Expected to connect, got %v", err)
+	}
+	defer nc.Close()
+	os.Remove(chainedFile)
+	ts.Shutdown()
+
+	ts = runTrustServer()
+	defer ts.Shutdown()
+
+	select {
+	case err := <-errs:
+		if !strings.Contains(err.Error(), "no such file or directory") {
+			t.Fatalf("Expected error about missing creds file, got %q", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Did not get error about missing creds file")
+	}
+}
+
 func TestUserJWTAndSeed(t *testing.T) {
 	if server.VERSION[0] == '1' {
 		t.Skip()
diff --git a/test/conn_test.go b/test/conn_test.go
index 777cb6f..995b3d1 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -374,7 +374,7 @@ func TestClientCertificateReloadOnServerRestart(t *testing.T) {
 	// wait for reconnection error (bad certificate)
 	select {
 	case err := <-errChan:
-		if !strings.Contains(err.Error(), "bad certificate") {
+		if !strings.Contains(err.Error(), "bad certificate") && !strings.Contains(err.Error(), "certificate required") {
 			t.Fatalf("Expected bad certificate error; got: %s", err)
 		}
 	case <-time.After(5 * time.Second):
diff --git a/test/js_test.go b/test/js_test.go
index d9ad8ca..d7ac093 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1124,6 +1124,255 @@ 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)
+	}
+	defer sub.Unsubscribe()
+	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 invalid arg error; got: %v", err)
+	}
+
+	// heartbeat value invalid
+	_, err = sub.Fetch(5, nats.PullHeartbeat(-1))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected invalid arg error; got: %v", err)
+	}
+
+	// set short timeout on JetStream context
+	js, err = nc.JetStream(nats.MaxWait(100 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sub1, err := js.PullSubscribe("foo", "")
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	defer sub.Unsubscribe()
+
+	// should produce invalid arg error based on default timeout from JetStream context
+	_, err = sub1.Fetch(5, nats.PullHeartbeat(100*time.Millisecond))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected invalid arg error; got: %v", err)
+	}
+
+	// overwrite default timeout with context timeout, fetch available messages
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	msgs, err = sub1.Fetch(10, nats.PullHeartbeat(100*time.Millisecond), nats.Context(ctx))
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	if len(msgs) != 5 {
+		t.Fatalf("Expected %d messages; got: %d", 5, len(msgs))
+	}
+	for _, msg := range msgs {
+		msg.Ack()
+	}
+
+	// overwrite default timeout with max wait, should time out because no messages are available
+	_, err = sub1.Fetch(5, nats.PullHeartbeat(100*time.Millisecond), nats.MaxWait(300*time.Millisecond))
+	if !errors.Is(err, nats.ErrTimeout) {
+		t.Fatalf("Expected timeout 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)
+	}
+	defer sub.Unsubscribe()
+	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.FetchBatch(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.FetchBatch(5, nats.PullHeartbeat(-1))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected no heartbeat error; got: %v", err)
+	}
+
+	// set short timeout on JetStream context
+	js, err = nc.JetStream(nats.MaxWait(100 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sub1, err := js.PullSubscribe("foo", "")
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	defer sub.Unsubscribe()
+
+	// should produce invalid arg error based on default timeout from JetStream context
+	_, err = sub1.Fetch(5, nats.PullHeartbeat(100*time.Millisecond))
+	if !errors.Is(err, nats.ErrInvalidArg) {
+		t.Fatalf("Expected invalid arg error; got: %v", err)
+	}
+
+	// overwrite default timeout with context timeout, fetch available messages
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+	defer cancel()
+	msgs, err = sub1.FetchBatch(10, nats.PullHeartbeat(100*time.Millisecond), nats.Context(ctx))
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	for msg := range msgs.Messages() {
+		msg.Ack()
+	}
+
+	// overwrite default timeout with max wait, should time out because no messages are available
+	msgs, err = sub1.FetchBatch(5, nats.PullHeartbeat(100*time.Millisecond), nats.MaxWait(300*time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %s", err)
+	}
+	<-msgs.Done()
+	if msgs.Error() != nil {
+		t.Fatalf("Unexpected error: %s", msgs.Error())
+	}
+}
+
 func TestPullSubscribeFetchBatch(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)
@@ -1899,14 +2148,46 @@ func TestJetStreamManagement(t *testing.T) {
 
 	// Create the stream using our client API.
 	var si *nats.StreamInfo
+
 	t.Run("create stream", func(t *testing.T) {
-		si, err := js.AddStream(&nats.StreamConfig{Name: "foo", Subjects: []string{"foo", "bar"}})
+		consLimits := nats.StreamConsumerLimits{
+			MaxAckPending:     100,
+			InactiveThreshold: 10 * time.Second,
+		}
+		cfg := &nats.StreamConfig{
+			Name:        "foo",
+			Subjects:    []string{"foo", "bar", "baz"},
+			Compression: nats.S2Compression,
+			ConsumerLimits: nats.StreamConsumerLimits{
+				MaxAckPending:     100,
+				InactiveThreshold: 10 * time.Second,
+			},
+			FirstSeq: 22,
+			Metadata: map[string]string{
+				"foo": "bar",
+				"baz": "quux",
+			},
+		}
+
+		si, err := js.AddStream(cfg)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
 		if si == nil || si.Config.Name != "foo" {
 			t.Fatalf("StreamInfo is not correct %+v", si)
 		}
+		if !reflect.DeepEqual(si.Config.Metadata, map[string]string{"foo": "bar", "baz": "quux"}) {
+			t.Fatalf("Metadata is not correct %+v", si.Config.Metadata)
+		}
+		if si.Config.Compression != nats.S2Compression {
+			t.Fatalf("Compression is not correct %+v", si.Config.Compression)
+		}
+		if si.Config.FirstSeq != 22 {
+			t.Fatalf("FirstSeq is not correct %+v", si.Config.FirstSeq)
+		}
+		if si.Config.ConsumerLimits != consLimits {
+			t.Fatalf("ConsumerLimits is not correct %+v", si.Config.ConsumerLimits)
+		}
 	})
 
 	t.Run("stream with given name already exists", func(t *testing.T) {
@@ -1991,7 +2272,14 @@ func TestJetStreamManagement(t *testing.T) {
 				t.Fatalf("Unexpected error: %v", err)
 			}
 			defer sub.Unsubscribe()
-			ci, err := js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy})
+			ci, err := js.AddConsumer("foo", &nats.ConsumerConfig{
+				Durable:   "dlc",
+				AckPolicy: nats.AckExplicitPolicy,
+				Metadata: map[string]string{
+					"foo": "bar",
+					"baz": "quux",
+				},
+			})
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
@@ -2005,6 +2293,9 @@ func TestJetStreamManagement(t *testing.T) {
 			if ci == nil || ci.Name != "dlc" || ci.Stream != "foo" {
 				t.Fatalf("ConsumerInfo is not correct %+v", ci)
 			}
+			if !reflect.DeepEqual(ci.Config.Metadata, map[string]string{"foo": "bar", "baz": "quux"}) {
+				t.Fatalf("Metadata is not correct %+v", ci.Config.Metadata)
+			}
 		})
 		t.Run("with name set", func(t *testing.T) {
 			sub, err := nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo.dlc-1")
@@ -2124,6 +2415,82 @@ func TestJetStreamManagement(t *testing.T) {
 			}
 		})
 
+		t.Run("durable consumer with multiple filter subjects", func(t *testing.T) {
+			sub, err := nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo.dlc-5")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer sub.Unsubscribe()
+			ci, err := js.AddConsumer("foo", &nats.ConsumerConfig{
+				Durable:        "dlc-5",
+				AckPolicy:      nats.AckExplicitPolicy,
+				FilterSubjects: []string{"foo", "bar"},
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			msg, err := sub.NextMsg(1 * time.Second)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if !strings.Contains(string(msg.Data), `"durable_name":"dlc-5"`) {
+				t.Fatalf("create consumer message is not correct: %q", string(msg.Data))
+			}
+			if ci == nil || ci.Config.Durable != "dlc-5" || !reflect.DeepEqual(ci.Config.FilterSubjects, []string{"foo", "bar"}) {
+				t.Fatalf("ConsumerInfo is not correct %+v", ci)
+			}
+		})
+
+		t.Run("ephemeral consumer with multiple filter subjects", func(t *testing.T) {
+			sub, err := nc.SubscribeSync("$JS.API.CONSUMER.CREATE.foo")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer sub.Unsubscribe()
+			ci, err := js.AddConsumer("foo", &nats.ConsumerConfig{
+				AckPolicy:      nats.AckExplicitPolicy,
+				FilterSubjects: []string{"foo", "bar"},
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			_, err = sub.NextMsg(1 * time.Second)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if ci == nil || !reflect.DeepEqual(ci.Config.FilterSubjects, []string{"foo", "bar"}) {
+				t.Fatalf("ConsumerInfo is not correct %+v", ci)
+			}
+		})
+
+		t.Run("multiple filter subjects errors", func(t *testing.T) {
+			// both filter subject and filter subjects provided
+			_, err := js.AddConsumer("foo", &nats.ConsumerConfig{
+				AckPolicy:      nats.AckExplicitPolicy,
+				FilterSubjects: []string{"foo", "bar"},
+				FilterSubject:  "baz",
+			})
+			if !errors.Is(err, nats.ErrDuplicateFilterSubjects) {
+				t.Fatalf("Expected: %v; got: %v", nats.ErrDuplicateFilterSubjects, err)
+			}
+			// overlapping filter subjects
+			_, err = js.AddConsumer("foo", &nats.ConsumerConfig{
+				AckPolicy:      nats.AckExplicitPolicy,
+				FilterSubjects: []string{"foo.*", "foo.A"},
+			})
+			if !errors.Is(err, nats.ErrOverlappingFilterSubjects) {
+				t.Fatalf("Expected: %v; got: %v", nats.ErrOverlappingFilterSubjects, err)
+			}
+			// empty filter subject in filter subjects
+			_, err = js.AddConsumer("foo", &nats.ConsumerConfig{
+				AckPolicy:      nats.AckExplicitPolicy,
+				FilterSubjects: []string{"foo", ""},
+			})
+			if !errors.Is(err, nats.ErrEmptyFilter) {
+				t.Fatalf("Expected: %v; got: %v", nats.ErrEmptyFilter, err)
+			}
+		})
+
 		t.Run("with invalid consumer name", func(t *testing.T) {
 			if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "test.durable"}); err != nats.ErrInvalidConsumerName {
 				t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidConsumerName, err)
@@ -2230,7 +2597,7 @@ func TestJetStreamManagement(t *testing.T) {
 		for info := range js.Consumers("foo") {
 			infos = append(infos, info)
 		}
-		if len(infos) != 6 || infos[0].Stream != "foo" {
+		if len(infos) != 8 || infos[0].Stream != "foo" {
 			t.Fatalf("ConsumerInfo is not correct %+v", infos)
 		}
 	})
@@ -2242,7 +2609,7 @@ func TestJetStreamManagement(t *testing.T) {
 		for name := range js.ConsumerNames("foo", nats.Context(ctx)) {
 			names = append(names, name)
 		}
-		if got, want := len(names), 6; got != want {
+		if got, want := len(names), 8; got != want {
 			t.Fatalf("Unexpected names, got=%d, want=%d", got, want)
 		}
 	})
@@ -2426,6 +2793,120 @@ func TestJetStreamManagement(t *testing.T) {
 	})
 }
 
+func TestStreamConfigMatches(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, js := jsClient(t, srv)
+	defer nc.Close()
+
+	cfg := nats.StreamConfig{
+		Name:                 "stream",
+		Description:          "desc",
+		Subjects:             []string{"foo.*"},
+		Retention:            nats.WorkQueuePolicy,
+		MaxConsumers:         10,
+		MaxMsgs:              100,
+		MaxBytes:             1000,
+		Discard:              nats.DiscardNew,
+		DiscardNewPerSubject: true,
+		MaxAge:               100 * time.Second,
+		MaxMsgsPerSubject:    1000,
+		MaxMsgSize:           10000,
+		Storage:              nats.MemoryStorage,
+		Replicas:             1,
+		NoAck:                true,
+		Duplicates:           10 * time.Second,
+		Sealed:               false,
+		DenyDelete:           true,
+		DenyPurge:            false,
+		AllowRollup:          true,
+		Compression:          nats.S2Compression,
+		FirstSeq:             5,
+		SubjectTransform:     &nats.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+		RePublish: &nats.RePublish{
+			Source:      ">",
+			Destination: "RP.>",
+			HeadersOnly: true,
+		},
+		AllowDirect: true,
+		ConsumerLimits: nats.StreamConsumerLimits{
+			InactiveThreshold: 10 * time.Second,
+			MaxAckPending:     500,
+		},
+		Metadata: map[string]string{
+			"foo": "bar",
+		},
+	}
+
+	s, err := js.AddStream(&cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(s.Config, cfg) {
+		t.Fatalf("StreamConfig doesn't match: %#v", s.Config)
+	}
+
+	cfgMirror := nats.StreamConfig{
+		Name:              "mirror",
+		MaxConsumers:      10,
+		MaxMsgs:           100,
+		MaxBytes:          1000,
+		MaxAge:            100 * time.Second,
+		MaxMsgsPerSubject: 1000,
+		MaxMsgSize:        10000,
+		Replicas:          1,
+		Duplicates:        10 * time.Second,
+		Mirror: &nats.StreamSource{
+			Name:        "stream",
+			OptStartSeq: 10,
+			SubjectTransforms: []nats.SubjectTransformConfig{
+				{Source: ">", Destination: "transformed.>"},
+			},
+		},
+		MirrorDirect:     true,
+		SubjectTransform: &nats.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+	}
+
+	s, err = js.AddStream(&cfgMirror)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(s.Config, cfgMirror) {
+		t.Fatalf("StreamConfig doesn't match: %#v", s.Config)
+	}
+
+	cfgSourcing := nats.StreamConfig{
+		Name:              "sourcing",
+		Subjects:          []string{"BAR"},
+		MaxConsumers:      10,
+		MaxMsgs:           100,
+		MaxBytes:          1000,
+		MaxAge:            100 * time.Second,
+		MaxMsgsPerSubject: 1000,
+		MaxMsgSize:        10000,
+		Replicas:          1,
+		Duplicates:        10 * time.Second,
+		Sources: []*nats.StreamSource{
+			{
+				Name:        "stream",
+				OptStartSeq: 10,
+				SubjectTransforms: []nats.SubjectTransformConfig{
+					{Source: ">", Destination: "transformed.>"},
+				},
+			},
+		},
+		SubjectTransform: &nats.SubjectTransformConfig{Source: ">", Destination: "transformed.>"},
+	}
+
+	s, err = js.AddStream(&cfgSourcing)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !reflect.DeepEqual(s.Config, cfgSourcing) {
+		t.Fatalf("StreamConfig doesn't match: %#v", s.Config)
+	}
+}
+
 func TestStreamLister(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -5401,6 +5882,70 @@ func TestJetStreamSubscribe_RateLimit(t *testing.T) {
 	}
 }
 
+func TestJetStreamSubscribe_FilterSubjects(t *testing.T) {
+	tests := []struct {
+		name    string
+		durable string
+	}{
+		{
+			name: "ephemeral consumer",
+		},
+		{
+			name:    "durable consumer",
+			durable: "cons",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer shutdownJSServerAndRemoveStorage(t, s)
+
+			nc, js := jsClient(t, s)
+			defer nc.Close()
+
+			var err error
+
+			_, err = js.AddStream(&nats.StreamConfig{
+				Name:     "TEST",
+				Subjects: []string{"foo", "bar", "baz"},
+			})
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			for i := 0; i < 5; i++ {
+				js.Publish("foo", []byte("msg"))
+			}
+			for i := 0; i < 5; i++ {
+				js.Publish("bar", []byte("msg"))
+			}
+			for i := 0; i < 5; i++ {
+				js.Publish("baz", []byte("msg"))
+			}
+
+			opts := []nats.SubOpt{nats.BindStream("TEST"), nats.ConsumerFilterSubjects("foo", "baz")}
+			if test.durable != "" {
+				opts = append(opts, nats.Durable(test.durable))
+			}
+			sub, err := js.SubscribeSync("", opts...)
+			if err != nil {
+				t.Fatalf("Unexpected error: %s", err)
+			}
+
+			for i := 0; i < 10; i++ {
+				msg, err := sub.NextMsg(500 * time.Millisecond)
+				if err != nil {
+					t.Fatalf("Unexpected error: %s", err)
+				}
+				if msg.Subject != "foo" && msg.Subject != "baz" {
+					t.Fatalf("Unexpected message subject: %s", msg.Subject)
+				}
+			}
+		})
+	}
+
+}
+
 func TestJetStreamSubscribe_ConfigCantChange(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)
@@ -5930,17 +6475,6 @@ func testJetStreamMirror_Source(t *testing.T, nodes ...*jsServer) {
 		}
 	})
 
-	t.Run("bind to stream with wrong subject fails", func(t *testing.T) {
-		_, err := js.SubscribeSync("secret", nats.BindStream("origin"))
-		if err == nil {
-			t.Fatal("Unexpected success")
-		}
-		apiErr := &nats.APIError{}
-		if !errors.As(err, &apiErr) || apiErr.ErrorCode != 10093 {
-			t.Fatalf("Expected API error 10093; got: %v", err)
-		}
-	})
-
 	t.Run("bind to origin stream", func(t *testing.T) {
 		// This would only avoid the stream names lookup.
 		sub, err := js.SubscribeSync("origin", nats.BindStream("origin"))
@@ -6094,103 +6628,78 @@ func testJetStreamMirror_Source(t *testing.T, nodes ...*jsServer) {
 		}
 	})
 
-	// Commenting out this test until we figure out what was the intent.
-	// Since v2.8.0, this test would fail with a "detected cycle" error,
-	// I guess because "m1" already sources "origin", so creating a
-	// stream with both as a source is bad.
-	/*
-		t.Run("create sourced stream from origin", func(t *testing.T) {
-			sources := make([]*nats.StreamSource, 0)
-			sources = append(sources, &nats.StreamSource{Name: "origin"})
-			sources = append(sources, &nats.StreamSource{Name: "m1"})
-			streamName := "s2"
-			_, err = js.AddStream(&nats.StreamConfig{
-				Name:     streamName,
-				Sources:  sources,
-				Storage:  nats.FileStorage,
-				Replicas: 1,
-			})
-			if err != nil {
-				t.Fatalf("Unexpected error creating stream: %v", err)
-			}
-
-			msgs := make([]*nats.RawStreamMsg, 0)
-
-			// Stored message sequences start at 1
-			startSequence := 1
-			expectedTotal := totalMsgs * 2
+	t.Run("bind to stream with subject not in stream", func(t *testing.T) {
+		// The Stream does not have a subject called 'nothing' but client is still able
+		// to bind to the origin stream even though it cannot consume messages.
+		// After updating the stream with the subject this consumer will be able to
+		// match and receive messages.
+		sub, err := js.SubscribeSync("nothing", nats.BindStream("origin"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, err = sub.NextMsg(1 * time.Second)
+		if !errors.Is(err, nats.ErrTimeout) {
+			t.Fatal("Expected timeout error")
+		}
 
-		GetNextMsg:
-			for i := startSequence; i < expectedTotal+1; i++ {
-				var (
-					err     error
-					seq     = uint64(i)
-					msg     *nats.RawStreamMsg
-					timeout = time.Now().Add(5 * time.Second)
-				)
+		info, err := sub.ConsumerInfo()
+		if err != nil {
+			t.Fatal(err)
+		}
+		got := info.Stream
+		expected := "origin"
+		if got != expected {
+			t.Fatalf("Expected %v, got %v", expected, got)
+		}
 
-			Retry:
-				for time.Now().Before(timeout) {
-					msg, err = js.GetMsg(streamName, seq)
-					if err != nil {
-						time.Sleep(100 * time.Millisecond)
-						continue Retry
-					}
-					msgs = append(msgs, msg)
-					continue GetNextMsg
-				}
-				if err != nil {
-					t.Fatalf("Unexpected error fetching seq=%v: %v", seq, err)
-				}
-			}
+		got = info.Config.FilterSubject
+		expected = "nothing"
+		if got != expected {
+			t.Fatalf("Expected %v, got %v", expected, got)
+		}
 
-			got := len(msgs)
-			if got < expectedTotal {
-				t.Errorf("Expected %v, got: %v", expectedTotal, got)
-			}
+		t.Run("can consume after stream update", func(t *testing.T) {
+			_, err = js.UpdateStream(&nats.StreamConfig{
+				Name: "origin",
+				Placement: &nats.Placement{
+					Tags: []string{"NODE_0"},
+				},
+				Storage:  nats.MemoryStorage,
+				Replicas: 1,
+				Subjects: []string{"origin", "nothing"},
+			})
+			js.Publish("nothing", []byte("hello world"))
 
-			si, err := js.StreamInfo(streamName)
+			msg, err := sub.NextMsg(1 * time.Second)
 			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			got = int(si.State.Msgs)
-			if got != expectedTotal {
-				t.Errorf("Expected %v, got: %v", expectedTotal, got)
+				t.Error(err)
 			}
-
-			got = len(si.Sources)
-			expected := 2
+			got = msg.Subject
+			expected = "nothing"
 			if got != expected {
-				t.Errorf("Expected %v, got: %v", expected, got)
+				t.Fatalf("Expected %v, got %v", expected, got)
 			}
 
-			t.Run("consume from sourced stream", func(t *testing.T) {
-				sub, err := js.SubscribeSync("origin", nats.BindStream(streamName))
-				if err != nil {
-					t.Fatal(err)
-				}
+		})
+	})
 
-				mmsgs := make([]*nats.Msg, 0)
-				for i := 0; i < totalMsgs; i++ {
-					msg, err := sub.NextMsg(2 * time.Second)
-					if err != nil {
-						t.Error(err)
-					}
-					meta, err := msg.Metadata()
-					if err != nil {
-						t.Error(err)
-					}
-					if meta.Stream != streamName {
-						t.Errorf("Expected m1, got: %v", meta.Stream)
-					}
-					mmsgs = append(mmsgs, msg)
-				}
-				if len(mmsgs) != totalMsgs {
-					t.Errorf("Expected to consume %v msgs, got: %v", totalMsgs, len(mmsgs))
-				}
-			})
+	t.Run("create sourced stream with a cycle", func(t *testing.T) {
+		// Since v2.8.0, this test would fail with a "detected cycle" error.
+		sources := make([]*nats.StreamSource, 0)
+		sources = append(sources, &nats.StreamSource{Name: "origin"})
+		sources = append(sources, &nats.StreamSource{Name: "m1"})
+		streamName := "s2"
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:     streamName,
+			Sources:  sources,
+			Storage:  nats.FileStorage,
+			Replicas: 1,
 		})
-	*/
+		var aerr *nats.APIError
+		if ok := errors.As(err, &aerr); !ok || aerr.ErrorCode != nats.JSStreamInvalidConfig {
+			t.Fatalf("Expected nats.APIError, got %v", err)
+		}
+	})
 }
 
 func TestJetStream_ClusterMultipleSubscribe(t *testing.T) {
diff --git a/test/object_test.go b/test/object_test.go
index 93e345e..28b1090 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -401,27 +401,45 @@ func TestObjectMetadata(t *testing.T) {
 	nc, js := jsClient(t, s)
 	defer nc.Close()
 
-	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "META-TEST"})
+	bucketMetadata := map[string]string{"foo": "bar", "baz": "boo"}
+	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{
+		Bucket:   "META-TEST",
+		Metadata: bucketMetadata,
+	})
+	expectOk(t, err)
+	status, err := obs.Status()
 	expectOk(t, err)
+	if !reflect.DeepEqual(status.Metadata(), bucketMetadata) {
+		t.Fatalf("invalid bucket metadata: %+v", status.Metadata())
+	}
 
 	// Simple with no Meta.
 	_, err = obs.PutString("A", "AAA")
 	expectOk(t, err)
-	_, err = obs.PutString("C", "CCC")
+	buf := bytes.NewBufferString("CCC")
+	objectMetadata := map[string]string{"name": "C", "description": "descC"}
+	info, err := obs.Put(&nats.ObjectMeta{Name: "C", Metadata: objectMetadata}, buf)
 	expectOk(t, err)
+	if !reflect.DeepEqual(info.Metadata, objectMetadata) {
+		t.Fatalf("invalid object metadata: %+v", info.Metadata)
+	}
 
 	meta := &nats.ObjectMeta{Name: "A"}
 	meta.Description = "descA"
 	meta.Headers = make(nats.Header)
 	meta.Headers.Set("color", "blue")
+	objectMetadata["description"] = "updated desc"
+	objectMetadata["version"] = "0.1"
+	meta.Metadata = objectMetadata
 
 	// simple update that does not change the name, just adds data
 	err = obs.UpdateMeta("A", meta)
 	expectOk(t, err)
 
-	info, err := obs.GetInfo("A")
+	info, err = obs.GetInfo("A")
 	expectOk(t, err)
-	if info.Name != "A" || info.Description != "descA" || info.Headers == nil || info.Headers.Get("color") != "blue" {
+	if info.Name != "A" || info.Description != "descA" || info.Headers == nil || info.Headers.Get("color") != "blue" ||
+		!reflect.DeepEqual(info.Metadata, objectMetadata) {
 		t.Fatalf("Update failed: %+v", info)
 	}
 
@@ -430,6 +448,7 @@ func TestObjectMetadata(t *testing.T) {
 	meta.Description = "descB"
 	meta.Headers = make(nats.Header)
 	meta.Headers.Set("color", "red")
+	meta.Metadata = nil
 
 	err = obs.UpdateMeta("A", meta)
 	expectOk(t, err)
@@ -441,7 +460,7 @@ func TestObjectMetadata(t *testing.T) {
 
 	info, err = obs.GetInfo("B")
 	expectOk(t, err)
-	if info.Name != "B" || info.Description != "descB" || info.Headers == nil || info.Headers.Get("color") != "red" {
+	if info.Name != "B" || info.Description != "descB" || info.Headers == nil || info.Headers.Get("color") != "red" || info.Metadata != nil {
 		t.Fatalf("Update failed: %+v", info)
 	}