Merge pull request #1453 from memphisdev/bugfix/RND-261-consumer-conf…

…ig-replica-count-exceeds-parent-stream bugfix/RND-261-consumer-config-replica-count-exceeds-parent-stream
superstreamlabs · Nov 28, 2023 · f00a73d · f00a73d
2 parents d5869d0 + 73717c8
commit f00a73d
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/server/memphis_handlers_monitoring.go b/server/memphis_handlers_monitoring.go
@@ -612,10 +612,10 @@ func (mh MonitoringHandler) GetStationOverviewData(c *gin.Context) {
 		messages, err = stationsHandler.GetMessagesFromPartition(station, fmt.Sprintf("%v$%v", stationName.Intern(), body.PartitionNumber), messagesToFetch, body.PartitionNumber)
 		if err != nil {
 			if IsNatsErr(err, JSStreamNotFoundErr) {
-				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetMessages: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetMessagesFromPartition: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
 				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
 			} else {
-				serv.Errorf("GetStationOverviewData at GetMessages: At station " + body.StationName + ": " + err.Error())
+				serv.Errorf("GetStationOverviewData at GetMessagesFromPartition: At station " + body.StationName + ": " + err.Error())
 				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
 			}
 			return

diff --git a/server/memphis_helper.go b/server/memphis_helper.go
@@ -1085,15 +1085,18 @@ func (s *Server) GetMessagesFromPartition(station models.Station, streamName str
 	if !station.IsNative {
 		filterSubj = ""
 	}
-
+	replicas := 1
+	if streamInfo.Config.Retention == InterestPolicy {
+		replicas = streamInfo.Config.Replicas
+	}
 	msgs, err := s.memphisGetMsgs(station.TenantName, filterSubj,
 		streamName,
 		startSequence,
 		messagesToFetch,
 		5*time.Second,
 		true,
 		station.RetentionType == "ack_based",
-		station.Replicas,
+		replicas,
 	)
 
 	if err != nil {
@@ -1180,15 +1183,14 @@ func getHdrLastIdxFromRaw(msg []byte) int {
 func (s *Server) memphisGetMsgs(tenantName, filterSubj, streamName string, startSeq uint64, amount int, timeout time.Duration, findHeader, isAckBasedStation bool, consumerReplicas int) ([]StoredMsg, error) {
 	uid, _ := uuid.NewV4()
 	durableName := "$memphis_fetch_messages_consumer_" + uid.String()
-	replicas := GetStationReplicas(1)
 
 	cc := ConsumerConfig{
 		FilterSubject: filterSubj,
 		OptStartSeq:   startSeq,
 		DeliverPolicy: DeliverByStartSequence,
 		Durable:       durableName,
 		AckPolicy:     AckExplicit,
-		Replicas:      replicas,
+		Replicas:      consumerReplicas,
 	}
 
 	err := s.memphisAddConsumer(tenantName, streamName, &cc)