Repository: nats-io/nats-server
Pull request #5245 (2024-03-26 08:10:09)

Add pedantic mode
This adds pedantic mode as requested in #4959 

Draft, as I need to make sure all cases are covered and tested.
I also want to check if the amount of changes in arguments passed to assigment functions could be reduced.

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>


ripienaar: Though more updates needed as all tests are failing, but if thats fixed then its good
Jarema: @derekcollison the reason behind `pedantic` being part of the request, not config, is that it should be scoped only to a given request, made as `pedantic` or not.
derekcollison: But that could still be easily part of the config no?

ripienaar: You make a pedantic API request not a pedantic mode stream. 

While the setting could be in the config what would happen in future - now we have to do config updates for non pedantic requests if initiallly it was or we need to have logic to ignore that specific config item only when storing to disk. 

The struct changes here also makes streams more consistent with consumers so kind of weren’t just for the sake of this one key 
ripienaar: Sorry ignore last part about consumers was thinking of another PR
derekcollison: This one ready for review and merge?
Jarema: @derekcollison yes. Just solved the conflicts and checking CU

Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 71e3e68e..61d6420b 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -119,9 +119,10 @@ type SequenceInfo struct {
 }
 
 type CreateConsumerRequest struct {
-	Stream string         `json:"stream_name"`
-	Config ConsumerConfig `json:"config"`
-	Action ConsumerAction `json:"action"`
+	Stream   string         `json:"stream_name"`
+	Config   ConsumerConfig `json:"config"`
+	Action   ConsumerAction `json:"action"`
+	Pedantic bool           `json:"pedantic,omitempty"`
 }
 
 type ConsumerAction int
@@ -437,7 +438,7 @@ const (
 )
 
 // Helper function to set consumer config defaults from above.
-func setConsumerConfigDefaults(config *ConsumerConfig, streamCfg *StreamConfig, lim *JSLimitOpts, accLim *JetStreamAccountLimits) {
+func setConsumerConfigDefaults(config *ConsumerConfig, streamCfg *StreamConfig, lim *JSLimitOpts, accLim *JetStreamAccountLimits, pedantic bool) *ApiError {
 	// Set to default if not specified.
 	if config.DeliverSubject == _EMPTY_ && config.MaxWaiting == 0 {
 		config.MaxWaiting = JSWaitQueueDefaultMax
@@ -452,12 +453,21 @@ func setConsumerConfigDefaults(config *ConsumerConfig, streamCfg *StreamConfig,
 	}
 	// If BackOff was specified that will override the AckWait and the MaxDeliver.
 	if len(config.BackOff) > 0 {
+		if pedantic && config.AckWait != config.BackOff[0] {
+			return NewJSPedanticError(errors.New("first backoff value has to equal batch AckWait"))
+		}
 		config.AckWait = config.BackOff[0]
 	}
 	if config.MaxAckPending == 0 {
+		if pedantic && streamCfg.ConsumerLimits.MaxAckPending > 0 {
+			return NewJSPedanticError(errors.New("max_ack_pending must be set if it's configured in stream limits"))
+		}
 		config.MaxAckPending = streamCfg.ConsumerLimits.MaxAckPending
 	}
 	if config.InactiveThreshold == 0 {
+		if pedantic && streamCfg.ConsumerLimits.InactiveThreshold > 0 {
+			return NewJSPedanticError(errors.New("inactive_threshold must be set if it's configured in stream limits"))
+		}
 		config.InactiveThreshold = streamCfg.ConsumerLimits.InactiveThreshold
 	}
 	// Set proper default for max ack pending if we are ack explicit and none has been set.
@@ -473,8 +483,12 @@ func setConsumerConfigDefaults(config *ConsumerConfig, streamCfg *StreamConfig,
 	}
 	// if applicable set max request batch size
 	if config.DeliverSubject == _EMPTY_ && config.MaxRequestBatch == 0 && lim.MaxRequestBatch > 0 {
+		if pedantic {
+			return NewJSPedanticError(errors.New("max_request_batch must be set if it's JetStream limits are set"))
+		}
 		config.MaxRequestBatch = lim.MaxRequestBatch
 	}
+	return nil
 }
 
 // Check the consumer config. If we are recovering don't check filter subjects.
@@ -702,15 +716,15 @@ func checkConsumerCfg(
 	return nil
 }
 
-func (mset *stream) addConsumerWithAction(config *ConsumerConfig, action ConsumerAction) (*consumer, error) {
-	return mset.addConsumerWithAssignment(config, _EMPTY_, nil, false, action)
+func (mset *stream) addConsumerWithAction(config *ConsumerConfig, action ConsumerAction, pedantic bool) (*consumer, error) {
+	return mset.addConsumerWithAssignment(config, _EMPTY_, nil, false, action, pedantic)
 }
 
 func (mset *stream) addConsumer(config *ConsumerConfig) (*consumer, error) {
-	return mset.addConsumerWithAction(config, ActionCreateOrUpdate)
+	return mset.addConsumerWithAction(config, ActionCreateOrUpdate, false)
 }
 
-func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname string, ca *consumerAssignment, isRecovering bool, action ConsumerAction) (*consumer, error) {
+func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname string, ca *consumerAssignment, isRecovering bool, action ConsumerAction, pedantic bool) (*consumer, error) {
 	// Check if this stream has closed.
 	if mset.closed.Load() {
 		return nil, NewJSStreamInvalidError()
@@ -740,8 +754,11 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	// Make sure we have sane defaults. Do so with the JS lock, otherwise a
 	// badly timed meta snapshot can result in a race condition.
 	mset.js.mu.Lock()
-	setConsumerConfigDefaults(config, &cfg, srvLim, selectedLimits)
+	err := setConsumerConfigDefaults(config, &cfg, srvLim, selectedLimits, pedantic)
 	mset.js.mu.Unlock()
+	if err != nil {
+		return nil, err
+	}
 
 	if err := checkConsumerCfg(config, srvLim, &cfg, acc, selectedLimits, isRecovering); err != nil {
 		return nil, err
diff --git a/server/errors.json b/server/errors.json
index 91c70bdf..c7eaef29 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1548,5 +1548,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSPedanticErrF",
+    "code": 400,
+    "error_code": 10157,
+    "description": "pedantic mode: {err}",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
diff --git a/server/jetstream.go b/server/jetstream.go
index 7627417d..3a986ccc 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1442,7 +1442,7 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 				// the consumer can reconnect. We will create it as a durable and switch it.
 				cfg.ConsumerConfig.Durable = ofi.Name()
 			}
-			obs, err := e.mset.addConsumerWithAssignment(&cfg.ConsumerConfig, _EMPTY_, nil, true, ActionCreateOrUpdate)
+			obs, err := e.mset.addConsumerWithAssignment(&cfg.ConsumerConfig, _EMPTY_, nil, true, ActionCreateOrUpdate, false)
 			if err != nil {
 				s.Warnf("    Error adding consumer %q: %v", cfg.Name, err)
 				continue
@@ -2603,7 +2603,7 @@ func (a *Account) addStreamTemplate(tc *StreamTemplateConfig) (*streamTemplate,
 	// FIXME(dlc) - Hacky
 	tcopy := tc.deepCopy()
 	tcopy.Config.Name = "_"
-	cfg, apiErr := s.checkStreamCfg(tcopy.Config, a)
+	cfg, apiErr := s.checkStreamCfg(tcopy.Config, a, false)
 	if apiErr != nil {
 		return nil, apiErr
 	}
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 066ebb7f..30b1fa5d 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1415,7 +1415,7 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	var cfg StreamConfig
+	var cfg StreamConfigRequest
 	if err := json.Unmarshal(msg, &cfg); err != nil {
 		resp.Error = NewJSInvalidJSONError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1456,13 +1456,13 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	if err := acc.jsNonClusteredStreamLimitsCheck(&cfg); err != nil {
+	if err := acc.jsNonClusteredStreamLimitsCheck(&cfg.StreamConfig); err != nil {
 		resp.Error = err
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
 
-	mset, err := acc.addStream(&cfg)
+	mset, err := acc.addStreamPedantic(&cfg.StreamConfig, cfg.Pedantic)
 	if err != nil {
 		if IsNatsErr(err, JSStreamStoreFailedF) {
 			s.Warnf("Stream create failed for '%s > %s': %v", acc, streamName, err)
@@ -1522,14 +1522,14 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		}
 		return
 	}
-	var ncfg StreamConfig
+	var ncfg StreamConfigRequest
 	if err := json.Unmarshal(msg, &ncfg); err != nil {
 		resp.Error = NewJSInvalidJSONError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
 
-	cfg, apiErr := s.checkStreamCfg(&ncfg, acc)
+	cfg, apiErr := s.checkStreamCfg(&ncfg.StreamConfig, acc, ncfg.Pedantic)
 	if apiErr != nil {
 		resp.Error = apiErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1546,7 +1546,7 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 	// Handle clustered version here.
 	if s.JetStreamIsClustered() {
 		// Always do in separate Go routine.
-		go s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, copyBytes(rmsg), &cfg, nil)
+		go s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, copyBytes(rmsg), &cfg, nil, ncfg.Pedantic)
 		return
 	}
 
@@ -1557,7 +1557,7 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	if err := mset.update(&cfg); err != nil {
+	if err := mset.updatePedantic(&cfg, ncfg.Pedantic); err != nil {
 		resp.Error = NewJSStreamUpdateError(err, Unless(err))
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
@@ -2578,7 +2578,8 @@ func (s *Server) jsLeaderServerStreamMoveRequest(sub *subscription, c *client, _
 		streamName, accName, cfg.Replicas, s.peerSetToNames(currPeers), s.peerSetToNames(peers))
 
 	// We will always have peers and therefore never do a callout, therefore it is safe to call inline
-	s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
+	// We should be fine ignoring pedantic mode here. as we do not touch configuration.
+	s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers, false)
 }
 
 // Request to have the metaleader move a stream on a peer to another
@@ -2684,7 +2685,7 @@ func (s *Server) jsLeaderServerStreamCancelMoveRequest(sub *subscription, c *cli
 		cfg.Replicas, streamName, accName, s.peerSetToNames(currPeers), s.peerSetToNames(peers))
 
 	// We will always have peers and therefore never do a callout, therefore it is safe to call inline
-	s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
+	s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers, false)
 }
 
 // Request to have an account purged
@@ -3373,7 +3374,7 @@ func (s *Server) jsStreamRestoreRequest(sub *subscription, c *client, _ *Account
 	}
 
 	// check stream config at the start of the restore process, not at the end
-	cfg, apiErr := s.checkStreamCfg(&req.Config, acc)
+	cfg, apiErr := s.checkStreamCfg(&req.Config, acc, false)
 	if apiErr != nil {
 		resp.Error = apiErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -4006,9 +4007,9 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 		// during this call, so place in Go routine to not block client.
 		// Router and Gateway API calls already in separate context.
 		if c.kind != ROUTER && c.kind != GATEWAY {
-			go s.jsClusteredConsumerRequest(ci, acc, subject, reply, rmsg, req.Stream, &req.Config, req.Action)
+			go s.jsClusteredConsumerRequest(ci, acc, subject, reply, rmsg, req.Stream, &req.Config, req.Action, req.Pedantic)
 		} else {
-			s.jsClusteredConsumerRequest(ci, acc, subject, reply, rmsg, req.Stream, &req.Config, req.Action)
+			s.jsClusteredConsumerRequest(ci, acc, subject, reply, rmsg, req.Stream, &req.Config, req.Action, req.Pedantic)
 		}
 		return
 	}
@@ -4033,7 +4034,7 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 		req.Config.PauseUntil = o.cfg.PauseUntil
 	}
 
-	o, err := stream.addConsumerWithAction(&req.Config, req.Action)
+	o, err := stream.addConsumerWithAction(&req.Config, req.Action, req.Pedantic)
 
 	if err != nil {
 		if IsNatsErr(err, JSConsumerStoreFailedErrF) {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index c46f2900..28bfea7d 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -3655,7 +3655,7 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 		mset.setStreamAssignment(sa)
 
 		// Call update.
-		if err = mset.updateWithAdvisory(cfg, !recovering); err != nil {
+		if err = mset.updateWithAdvisory(cfg, !recovering, false); err != nil {
 			s.Warnf("JetStream cluster error updating stream %q for account %q: %v", cfg.Name, acc.Name, err)
 		}
 	}
@@ -3800,7 +3800,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 			mset.setStreamAssignment(sa)
 			// Check if our config has really been updated.
 			if !reflect.DeepEqual(mset.config(), sa.Config) {
-				if err = mset.updateWithAdvisory(sa.Config, false); err != nil {
+				if err = mset.updateWithAdvisory(sa.Config, false, false); err != nil {
 					s.Warnf("JetStream cluster error updating stream %q for account %q: %v", sa.Config.Name, acc.Name, err)
 					if osa != nil {
 						// Process the raft group and make sure it's running if needed.
@@ -3819,7 +3819,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 			}
 		} else if err == NewJSStreamNotFoundError() {
 			// Add in the stream here.
-			mset, err = acc.addStreamWithAssignment(sa.Config, nil, sa)
+			mset, err = acc.addStreamWithAssignment(sa.Config, nil, sa, false)
 		}
 		if mset != nil {
 			mset.setCreatedTime(sa.Created)
@@ -4347,7 +4347,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 	var didCreate, isConfigUpdate, needsLocalResponse bool
 	if o == nil {
 		// Add in the consumer if needed.
-		if o, err = mset.addConsumerWithAssignment(ca.Config, ca.Name, ca, js.isMetaRecovering(), ActionCreateOrUpdate); err == nil {
+		if o, err = mset.addConsumerWithAssignment(ca.Config, ca.Name, ca, js.isMetaRecovering(), ActionCreateOrUpdate, false); err == nil {
 			didCreate = true
 		}
 	} else {
@@ -6069,7 +6069,7 @@ func (js *jetStream) jsClusteredStreamLimitsCheck(acc *Account, cfg *StreamConfi
 	return nil
 }
 
-func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, config *StreamConfig) {
+func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, config *StreamConfigRequest) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -6077,7 +6077,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 
 	var resp = JSApiStreamCreateResponse{ApiResponse: ApiResponse{Type: JSApiStreamCreateResponseType}}
 
-	ccfg, apiErr := s.checkStreamCfg(config, acc)
+	ccfg, apiErr := s.checkStreamCfg(&config.StreamConfig, acc, config.Pedantic)
 	if apiErr != nil {
 		resp.Error = apiErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -6221,7 +6221,7 @@ func sysRequest[T any](s *Server, subjFormat string, args ...any) (*T, error) {
 	}
 }
 
-func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig, peerSet []string) {
+func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig, peerSet []string, pedantic bool) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -6247,7 +6247,7 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	var newCfg *StreamConfig
 	if jsa := js.accounts[acc.Name]; jsa != nil {
 		js.mu.Unlock()
-		ncfg, err := jsa.configUpdateCheck(osa.Config, cfg, s)
+		ncfg, err := jsa.configUpdateCheck(osa.Config, cfg, s, pedantic)
 		js.mu.Lock()
 		if err != nil {
 			resp.Error = NewJSStreamUpdateError(err, Unless(err))
@@ -7163,7 +7163,7 @@ func (cc *jetStreamCluster) createGroupForConsumer(cfg *ConsumerConfig, sa *stre
 }
 
 // jsClusteredConsumerRequest is first point of entry to create a consumer in clustered mode.
-func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, stream string, cfg *ConsumerConfig, action ConsumerAction) {
+func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, stream string, cfg *ConsumerConfig, action ConsumerAction, pedantic bool) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -7185,7 +7185,11 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 	}
 	srvLim := &s.getOpts().JetStreamLimits
 	// Make sure we have sane defaults
-	setConsumerConfigDefaults(cfg, &streamCfg, srvLim, selectedLimits)
+	if err := setConsumerConfigDefaults(cfg, &streamCfg, srvLim, selectedLimits, pedantic); err != nil {
+		resp.Error = err
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
 
 	if err := checkConsumerCfg(cfg, srvLim, &streamCfg, acc, selectedLimits, false); err != nil {
 		resp.Error = err
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 277675fa..7115dc4a 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -284,6 +284,9 @@ const (
 	// JSNotEnabledForAccountErr JetStream not enabled for account
 	JSNotEnabledForAccountErr ErrorIdentifier = 10039
 
+	// JSPedanticErrF pedantic mode: {err}
+	JSPedanticErrF ErrorIdentifier = 10157
+
 	// JSPeerRemapErr peer remap failed
 	JSPeerRemapErr ErrorIdentifier = 10075
 
@@ -566,6 +569,7 @@ var (
 		JSNotEmptyRequestErr:                       {Code: 400, ErrCode: 10038, Description: "expected an empty request payload"},
 		JSNotEnabledErr:                            {Code: 503, ErrCode: 10076, Description: "JetStream not enabled"},
 		JSNotEnabledForAccountErr:                  {Code: 503, ErrCode: 10039, Description: "JetStream not enabled for account"},
+		JSPedanticErrF:                             {Code: 400, ErrCode: 10157, Description: "pedantic mode: {err}"},
 		JSPeerRemapErr:                             {Code: 503, ErrCode: 10075, Description: "peer remap failed"},
 		JSRaftGeneralErrF:                          {Code: 500, ErrCode: 10041, Description: "{err}"},
 		JSReplicasCountCannotBeNegative:            {Code: 400, ErrCode: 10133, Description: "replicas count cannot be negative"},
@@ -1667,6 +1671,22 @@ func NewJSNotEnabledForAccountError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSNotEnabledForAccountErr]
 }
 
+// NewJSPedanticError creates a new JSPedanticErrF error: "pedantic mode: {err}"
+func NewJSPedanticError(err error, opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	e := ApiErrors[JSPedanticErrF]
+	args := e.toReplacerArgs([]interface{}{"{err}", err})
+	return &ApiError{
+		Code:        e.Code,
+		ErrCode:     e.ErrCode,
+		Description: strings.NewReplacer(args...).Replace(e.Description),
+	}
+}
+
 // NewJSPeerRemapError creates a new JSPeerRemapErr error: "peer remap failed"
 func NewJSPeerRemapError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/stream.go b/server/stream.go
index 68cf0be4..e71cf4ce 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -36,6 +36,14 @@ import (
 	"github.com/nats-io/nuid"
 )
 
+// StreamConfigRequest is used to create or update a stream.
+type StreamConfigRequest struct {
+	StreamConfig
+	// This is not part of the StreamConfig, because its scoped to request,
+	// and not to the stream itself.
+	Pedantic bool `json:"pedantic,omitempty"`
+}
+
 // StreamConfig will determine the name, subjects and retention policy
 // for a given stream. If subjects is empty the name will be used.
 type StreamConfig struct {
@@ -393,15 +401,19 @@ const StreamMaxReplicas = 5
 
 // AddStream adds a stream for the given account.
 func (a *Account) addStream(config *StreamConfig) (*stream, error) {
-	return a.addStreamWithAssignment(config, nil, nil)
+	return a.addStreamWithAssignment(config, nil, nil, false)
 }
 
 // AddStreamWithStore adds a stream for the given account with custome store config options.
 func (a *Account) addStreamWithStore(config *StreamConfig, fsConfig *FileStoreConfig) (*stream, error) {
-	return a.addStreamWithAssignment(config, fsConfig, nil)
+	return a.addStreamWithAssignment(config, fsConfig, nil, false)
+}
+
+func (a *Account) addStreamPedantic(config *StreamConfig, pedantic bool) (*stream, error) {
+	return a.addStreamWithAssignment(config, nil, nil, pedantic)
 }
 
-func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileStoreConfig, sa *streamAssignment) (*stream, error) {
+func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileStoreConfig, sa *streamAssignment, pedantic bool) (*stream, error) {
 	s, jsa, err := a.checkForJetStream()
 	if err != nil {
 		return nil, err
@@ -415,7 +427,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	}
 
 	// Sensible defaults.
-	cfg, apiErr := s.checkStreamCfg(config, a)
+	cfg, apiErr := s.checkStreamCfg(config, a, pedantic)
 	if apiErr != nil {
 		return nil, apiErr
 	}
@@ -1169,7 +1181,7 @@ func (jsa *jsAccount) subjectsOverlap(subjects []string, self *stream) bool {
 // StreamDefaultDuplicatesWindow default duplicates window.
 const StreamDefaultDuplicatesWindow = 2 * time.Minute
 
-func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfig, *ApiError) {
+func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account, pedantic bool) (StreamConfig, *ApiError) {
 	lim := &s.getOpts().JetStreamLimits
 
 	if config == nil {
@@ -1226,9 +1238,15 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 	if cfg.Duplicates == 0 && cfg.Mirror == nil {
 		maxWindow := StreamDefaultDuplicatesWindow
 		if lim.Duplicates > 0 && maxWindow > lim.Duplicates {
+			if pedantic {
+				return StreamConfig{}, NewJSPedanticError(fmt.Errorf("pedantic mode: duplicate window limits are higher than current limits"))
+			}
 			maxWindow = lim.Duplicates
 		}
 		if cfg.MaxAge != 0 && cfg.MaxAge < maxWindow {
+			if pedantic {
+				return StreamConfig{}, NewJSPedanticError(fmt.Errorf("pedantic mode: duplicate window cannot be bigger than max age"))
+			}
 			cfg.Duplicates = cfg.MaxAge
 		} else {
 			cfg.Duplicates = maxWindow
@@ -1343,6 +1361,9 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 			}
 			// Determine if we are inheriting direct gets.
 			if exists, ocfg := getStream(cfg.Mirror.Name); exists {
+				if pedantic && cfg.MirrorDirect != ocfg.AllowDirect {
+					return StreamConfig{}, NewJSPedanticError(fmt.Errorf("origin stream has direct get set, mirror has it disabled"))
+				}
 				cfg.MirrorDirect = ocfg.AllowDirect
 			} else if js := s.getJetStream(); js != nil && js.isClustered() {
 				// Could not find it here. If we are clustered we can look it up.
@@ -1350,6 +1371,10 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 				if cc := js.cluster; cc != nil {
 					if as := cc.streams[acc.Name]; as != nil {
 						if sa := as[cfg.Mirror.Name]; sa != nil {
+							if pedantic && cfg.MirrorDirect != sa.Config.AllowDirect {
+								js.mu.RUnlock()
+								return StreamConfig{}, NewJSPedanticError(fmt.Errorf("origin stream has direct get set, mirror has it disabled"))
+							}
 							cfg.MirrorDirect = sa.Config.AllowDirect
 						}
 					}
@@ -1571,6 +1596,9 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		// Also make sure it does not form a cycle.
 		// Empty same as all.
 		if cfg.RePublish.Source == _EMPTY_ {
+			if pedantic {
+				return StreamConfig{}, NewJSPedanticError(fmt.Errorf("pedantic mode: republish source can not be empty"))
+			}
 			cfg.RePublish.Source = fwcs
 		}
 		var formsCycle bool
@@ -1621,8 +1649,8 @@ func (mset *stream) fileStoreConfig() (FileStoreConfig, error) {
 }
 
 // Do not hold jsAccount or jetStream lock
-func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server) (*StreamConfig, error) {
-	cfg, apiErr := s.checkStreamCfg(new, jsa.acc())
+func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server, pedantic bool) (*StreamConfig, error) {
+	cfg, apiErr := s.checkStreamCfg(new, jsa.acc(), pedantic)
 	if apiErr != nil {
 		return nil, apiErr
 	}
@@ -1680,6 +1708,7 @@ func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server) (*Str
 	}
 
 	// Do some adjustments for being sealed.
+	// Pedantic mode will allow those changes to be made, as they are determinictic and important to get a sealed stream.
 	if cfg.Sealed {
 		cfg.MaxAge = 0
 		cfg.Discard = DiscardNew
@@ -1748,11 +1777,15 @@ func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server) (*Str
 
 // Update will allow certain configuration properties of an existing stream to be updated.
 func (mset *stream) update(config *StreamConfig) error {
-	return mset.updateWithAdvisory(config, true)
+	return mset.updateWithAdvisory(config, true, false)
+}
+
+func (mset *stream) updatePedantic(config *StreamConfig, pedantic bool) error {
+	return mset.updateWithAdvisory(config, true, pedantic)
 }
 
 // Update will allow certain configuration properties of an existing stream to be updated.
-func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool) error {
+func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool, pedantic bool) error {
 	_, jsa, err := mset.acc.checkForJetStream()
 	if err != nil {
 		return err
@@ -1763,7 +1796,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 	s := mset.srv
 	mset.mu.RUnlock()
 
-	cfg, err := mset.jsa.configUpdateCheck(&ocfg, config, s)
+	cfg, err := mset.jsa.configUpdateCheck(&ocfg, config, s, pedantic)
 	if err != nil {
 		return NewJSStreamInvalidConfigError(err, Unless(err))
 	}
@@ -5976,7 +6009,7 @@ func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error
 		return nil, err
 	}
 
-	cfg, apiErr := s.checkStreamCfg(ncfg, a)
+	cfg, apiErr := s.checkStreamCfg(ncfg, a, false)
 	if apiErr != nil {
 		return nil, apiErr
 	}


Test patch:
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 46ce0851..611d5738 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -482,7 +482,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_NoError(t, err)
 	// Create consumer again. Should be ok if action is CREATE but config is exactly the same.
 	_, err = mset.addConsumerWithAction(&ConsumerConfig{
@@ -491,7 +491,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_NoError(t, err)
 	// Create consumer again. Should error if action is CREATE.
 	_, err = mset.addConsumerWithAction(&ConsumerConfig{
@@ -500,7 +500,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_Error(t, err)
 
 	// Update existing consumer. Should be fine, as consumer exists.
@@ -510,7 +510,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionUpdate)
+	}, ActionUpdate, false)
 	require_NoError(t, err)
 
 	// Update consumer. Should error, as this consumer does not exist.
@@ -520,7 +520,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionUpdate)
+	}, ActionUpdate, false)
 	require_Error(t, err)
 
 	// Create new ephemeral. Should be fine as the consumer doesn't exist already
@@ -530,7 +530,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_NoError(t, err)
 
 	// Trying to create it again right away. Should error as it already exists (and hasn't been cleaned up yet)
@@ -540,7 +540,7 @@ func TestJetStreamConsumerActions(t *testing.T) {
 		AckPolicy:      AckExplicit,
 		DeliverPolicy:  DeliverAll,
 		AckWait:        time.Second * 30,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_Error(t, err)
 }
 
@@ -563,21 +563,21 @@ func TestJetStreamConsumerActionsOnWorkQueuePolicyStream(t *testing.T) {
 		Durable:        "C1",
 		FilterSubjects: []string{"one", "two"},
 		AckPolicy:      AckExplicit,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_NoError(t, err)
 
 	_, err = mset.addConsumerWithAction(&ConsumerConfig{
 		Durable:        "C2",
 		FilterSubjects: []string{"three", "four"},
 		AckPolicy:      AckExplicit,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_NoError(t, err)
 
 	_, err = mset.addConsumerWithAction(&ConsumerConfig{
 		Durable:        "C3",
 		FilterSubjects: []string{"five.*"},
 		AckPolicy:      AckExplicit,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_NoError(t, err)
 
 	// Updating a consumer by removing a previous subject filter.
@@ -585,7 +585,7 @@ func TestJetStreamConsumerActionsOnWorkQueuePolicyStream(t *testing.T) {
 		Durable:        "C1",
 		FilterSubjects: []string{"one"}, // Remove a subject.
 		AckPolicy:      AckExplicit,
-	}, ActionUpdate)
+	}, ActionUpdate, false)
 	require_NoError(t, err)
 
 	// Updating a consumer without overlapping subjects.
@@ -593,7 +593,7 @@ func TestJetStreamConsumerActionsOnWorkQueuePolicyStream(t *testing.T) {
 		Durable:        "C2",
 		FilterSubjects: []string{"three", "four", "two"}, // Add previously removed subject.
 		AckPolicy:      AckExplicit,
-	}, ActionUpdate)
+	}, ActionUpdate, false)
 	require_NoError(t, err)
 
 	// Creating a consumer with overlapping subjects should return an error.
@@ -601,7 +601,7 @@ func TestJetStreamConsumerActionsOnWorkQueuePolicyStream(t *testing.T) {
 		Durable:        "C4",
 		FilterSubjects: []string{"one", "two", "three", "four"},
 		AckPolicy:      AckExplicit,
-	}, ActionCreate)
+	}, ActionCreate, false)
 	require_Error(t, err)
 	if !IsNatsErr(err, JSConsumerWQConsumerNotUniqueErr) {
 		t.Errorf("want error %q, got %q", ApiErrors[JSConsumerWQConsumerNotUniqueErr], err)
@@ -612,7 +612,7 @@ func TestJetStreamConsumerActionsOnWorkQueuePolicyStream(t *testing.T) {
 		Durable:        "C3",
 		FilterSubjects: []string{"one", "two", "three", "four"},
 		AckPolicy:      AckExplicit,
-	}, ActionUpdate)
+	}, ActionUpdate, false)
 	require_Error(t, err)
 	if !IsNatsErr(err, JSConsumerWQConsumerNotUniqueErr) {
 		t.Errorf("want error %q, got %q", ApiErrors[JSConsumerWQConsumerNotUniqueErr], err)
@@ -1224,6 +1224,184 @@ func TestJetStreamConsumerLongSubjectHang(t *testing.T) {
 	}
 }
 
+func TestJetStreamConsumerPedanticMode(t *testing.T) {
+
+	singleServerTemplate := `
+			listen: 127.0.0.1:-1
+			jetstream: {
+				max_mem_store: 2MB,
+				max_file_store: 8MB,
+				store_dir: '%s',
+				limits: {max_request_batch: 250}
+			}
+			no_auth_user: u
+			accounts {
+				ONE {
+					users = [ { user: "u", pass: "s3cr3t!" } ]
+					jetstream: enabled
+				}
+				$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+			}`
+
+	clusterTemplate := `
+			listen: 127.0.0.1:-1
+			server_name: %s
+			jetstream: {
+				max_mem_store: 2MB,
+				max_file_store: 8MB,
+				store_dir: '%s',
+				limits: {max_request_batch: 250}
+			}
+			cluster {
+				name: %s
+				listen: 127.0.0.1:%d
+				routes = [%s]
+			}
+			no_auth_user: u
+			accounts {
+				ONE {
+					users = [ { user: "u", pass: "s3cr3t!" } ]
+					jetstream: enabled
+				}
+				$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+			}`
+
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, _ := jsClientConnect(t, s)
+	defer nc.Close()
+
+	tests := []struct {
+		name                  string
+		givenConfig           ConsumerConfig
+		givenLimits           nats.StreamConsumerLimits
+		serverTemplateSingle  string
+		serverTemplateCluster string
+		shouldError           bool
+		pedantic              bool
+		replicas              int
+	}{
+		{
+			name: "default_non_pedantic",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+			},
+			givenLimits: nats.StreamConsumerLimits{
+				InactiveThreshold: time.Minute,
+				MaxAckPending:     100,
+			},
+			shouldError: false,
+			pedantic:    false,
+		},
+		{
+			name: "default_pedantic_inactive_threshold",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+			},
+			givenLimits: nats.StreamConsumerLimits{
+				InactiveThreshold: time.Minute,
+			},
+			shouldError: true,
+			pedantic:    true,
+		},
+		{
+			name: "default_pedantic_max_ack_pending",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+			},
+			givenLimits: nats.StreamConsumerLimits{
+				MaxAckPending: 100,
+			},
+			shouldError: true,
+			pedantic:    true,
+		},
+		{
+			name: "pedantic_backoff_no_ack_wait",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+				BackOff: []time.Duration{time.Second, time.Minute},
+			},
+			pedantic:    true,
+			shouldError: true,
+		},
+		{
+			name: "backoff_no_ack_wait",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+				BackOff: []time.Duration{time.Second, time.Minute},
+			},
+			pedantic:    false,
+			shouldError: false,
+		},
+		{
+			name: "max_batch_requests",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+			},
+			serverTemplateSingle:  singleServerTemplate,
+			serverTemplateCluster: clusterTemplate,
+			pedantic:              false,
+			shouldError:           false,
+		},
+		{
+			name: "pedantic_max_batch_requests",
+			givenConfig: ConsumerConfig{
+				Durable: "durable",
+			},
+			serverTemplateSingle:  singleServerTemplate,
+			serverTemplateCluster: clusterTemplate,
+			pedantic:              true,
+			shouldError:           true,
+		},
+	}
+
+	for _, test := range tests {
+		for _, mode := range []string{"clustered", "single"} {
+			t.Run(fmt.Sprintf("%v_%v", mode, test.name), func(t *testing.T) {
+
+				var s *Server
+				if mode == "single" {
+					s = RunBasicJetStreamServer(t)
+					defer s.Shutdown()
+				} else {
+					c := createJetStreamClusterExplicit(t, "R3S", 3)
+					defer c.shutdown()
+					s = c.randomServer()
+				}
+
+				replicas := 1
+				if mode == "clustered" {
+					replicas = 3
+				}
+
+				nc, js := jsClientConnect(t, s)
+				defer nc.Close()
+
+				js.AddStream(&nats.StreamConfig{
+					Name:     test.name,
+					Subjects: []string{"foo"},
+					Replicas: replicas,
+					ConsumerLimits: nats.StreamConsumerLimits{
+						InactiveThreshold: time.Minute,
+						MaxAckPending:     100,
+					},
+				})
+
+				_, err := addConsumerWithError(t, nc, &CreateConsumerRequest{
+					Stream:   test.name,
+					Config:   test.givenConfig,
+					Action:   ActionCreateOrUpdate,
+					Pedantic: test.pedantic,
+				})
+				require_True(t, (err != nil) == test.shouldError)
+				if err != nil {
+					require_True(t, strings.Contains(err.Error(), "pedantic"))
+				}
+			})
+		}
+	}
+}
 func Benchmark____JetStreamConsumerIsFilteredMatch(b *testing.B) {
 	subject := "foo.bar.do.not.match.any.filter.subject"
 	for n := 1; n <= 1024; n *= 2 {
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index 6ffc33e8..719ea06f 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -1617,6 +1617,36 @@ func (c *cluster) stableTotalSubs() (total int) {
 
 }
 
+func addStreamPedanticWithError(t *testing.T, nc *nats.Conn, cfg *StreamConfigRequest) (*StreamInfo, *ApiError) {
+	t.Helper()
+	req, err := json.Marshal(cfg)
+	require_NoError(t, err)
+	rmsg, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, 5*time.Second)
+	require_NoError(t, err)
+	var resp JSApiStreamCreateResponse
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	if resp.Type != JSApiStreamCreateResponseType {
+		t.Fatalf("Invalid response type %s expected %s", resp.Type, JSApiStreamCreateResponseType)
+	}
+	return resp.StreamInfo, resp.Error
+}
+
+func updateStreamPedanticWithError(t *testing.T, nc *nats.Conn, cfg *StreamConfigRequest) (*StreamInfo, *ApiError) {
+	t.Helper()
+	req, err := json.Marshal(cfg)
+	require_NoError(t, err)
+	rmsg, err := nc.Request(fmt.Sprintf(JSApiStreamUpdateT, cfg.Name), req, time.Second)
+	require_NoError(t, err)
+	var resp JSApiStreamCreateResponse
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	if resp.Type != JSApiStreamUpdateResponseType {
+		t.Fatalf("Invalid response type %s expected %s", resp.Type, JSApiStreamUpdateResponseType)
+	}
+	return resp.StreamInfo, resp.Error
+}
+
 func addStream(t *testing.T, nc *nats.Conn, cfg *StreamConfig) *StreamInfo {
 	t.Helper()
 	si, err := addStreamWithError(t, nc, cfg)
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 602bb437..8bb40f65 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -23954,3 +23954,117 @@ func TestJetStreamInterestStreamWithDuplicateMessages(t *testing.T) {
 	require_Equal(t, pa.Sequence, 1)
 	require_Equal(t, pa.Duplicate, true)
 }
+
+func TestJetStreamStreamCreatePedanticMode(t *testing.T) {
+	cfgFmt := []byte(fmt.Sprintf(`
+        jetstream: {
+            enabled: true
+            max_file_store: 100MB
+            store_dir: %s
+            limits: {duplicate_window: "1m", max_request_batch: 250}
+        }
+        accounts: {
+            myacc: {
+                jetstream: enabled
+                users: [ { user: user, password: pass  } ]
+            }
+        }
+        no_auth_user: user
+	`, t.TempDir()))
+
+	conf := createConfFile(t, cfgFmt)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, _ := jsClientConnect(t, s)
+	defer nc.Close()
+
+	tests := []struct {
+		name      string
+		cfg       StreamConfigRequest
+		shouldErr bool
+		update    bool
+	}{
+		{
+			name: "too_high_duplicate",
+			cfg: StreamConfigRequest{
+				StreamConfig: StreamConfig{
+					Name:       "TEST",
+					MaxAge:     time.Minute,
+					Duplicates: time.Hour,
+					Storage:    FileStorage,
+				},
+				Pedantic: true,
+			},
+			shouldErr: true,
+		},
+		{
+			name: "duplicate_over_limits",
+			cfg: StreamConfigRequest{
+				StreamConfig: StreamConfig{
+					Name:       "TEST",
+					MaxAge:     time.Hour * 60,
+					Duplicates: time.Hour,
+					Storage:    FileStorage,
+				},
+				Pedantic: false,
+			},
+			shouldErr: true,
+		},
+		{
+			name: "duplicate_window_within_limits",
+			cfg: StreamConfigRequest{
+				StreamConfig: StreamConfig{
+					Name:       "TEST",
+					MaxAge:     time.Hour * 60,
+					Duplicates: time.Second * 30,
+					Storage:    FileStorage,
+				},
+				Pedantic: false,
+			},
+			shouldErr: false,
+		},
+		{
+			name: "update_too_high_duplicate",
+			cfg: StreamConfigRequest{
+				StreamConfig: StreamConfig{
+					Name:       "TEST",
+					MaxAge:     time.Minute,
+					Duplicates: time.Hour,
+					Storage:    FileStorage,
+				},
+				Pedantic: true,
+			},
+			update:    true,
+			shouldErr: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if !test.update {
+				_, err := addStreamPedanticWithError(t, nc, &test.cfg)
+				require_True(t, (err != nil) == test.shouldErr)
+			} else {
+				_, err := updateStreamPedanticWithError(t, nc, &test.cfg)
+				require_True(t, (err != nil) == test.shouldErr)
+			}
+		})
+	}
+}
+
+func addConsumerWithError(t *testing.T, nc *nats.Conn, cfg *CreateConsumerRequest) (*ConsumerInfo, *ApiError) {
+	t.Helper()
+	req, err := json.Marshal(cfg)
+	require_NoError(t, err)
+	rmsg, err := nc.Request(fmt.Sprintf(JSApiDurableCreateT, cfg.Stream, cfg.Config.Durable), req, 5*time.Second)
+
+	require_NoError(t, err)
+	var resp JSApiConsumerCreateResponse
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	if resp.Type != JSApiConsumerCreateResponseType {
+		t.Fatalf("Invalid response type %s expected %s", resp.Type, JSApiConsumerCreateResponseType)
+	}
+	return resp.ConsumerInfo, resp.Error
+}