Only meta/stream leader responds to consumer list

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/jetstream_api.go

@@ -4650,6 +4650,32 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account,
 
 	// Clustered mode will invoke a scatter and gather.
 	if s.JetStreamIsClustered() {
+		// Determine if we should proceed here when we are in clustered mode.
+		js, cc := s.getJetStreamCluster()
+		if js == nil || cc == nil {
+			return
+		}
+		if js.isLeaderless() {
+			resp.Error = NewJSClusterNotAvailError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		isLeader := s.JetStreamIsLeader()
+		js.mu.RLock()
+		sa := js.streamAssignment(acc.Name, streamName)
+		if sa != nil && sa.Config != nil && sa.Config.ManagesConsumers {
+			// If the stream manages its own consumers then we care about the stream
+			// leader at this point, not the metaleader, as the metaleader will not
+			// know or not whether the consumer is meant to exist.
+			isLeader = cc.isStreamLeader(acc.Name, streamName)
+		}
+		js.mu.RUnlock()
+
+		// Make sure we are leader.
+		if !isLeader {
+			return
+		}
+
 		// Need to copy these off before sending.. don't move this inside startGoRoutine!!!
 		msg = copyBytes(msg)
 		s.startGoRoutine(func() {