gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-02 03:38:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

Make sure to return no consumer if we are still waiting on a leader to be elected.

Browse Source 
Signed-off-by: Derek Collison <derek@nats.io>
This commit is contained in:
Derek Collison
2021-04-21 12:13:10 -07:00
parent 3d928c847a
commit 9f1bc5882c
2 changed files with 13 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
server/jetstream_api.go
View File

@@ -1480,6 +1480,7 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, subject, rep
resp.Total = numStreams
resp.Limit = JSApiNamesLimit
resp.Offset = offset
s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
}
@@ -3281,6 +3282,10 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
js.mu.RLock()
isLeader, sa, ca := cc.isLeader(), js.streamAssignment(acc.Name, streamName), js.consumerAssignment(acc.Name, streamName, consumerName)
// Ignore pending consumers for now.
if ca != nil && ca.pending {
ca = nil
}
js.mu.RUnlock()
if isLeader && ca == nil {
@@ -3320,6 +3325,13 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
resp.Error = jsClusterNotAvailErr
s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
}
// Check here if we are a member and this is just a new consumer that does not have a leader yet.
if rg := ca.Group; rg.node != nil && rg.isMember(cc.meta.ID()) {
if rg.node.GroupLeader() == _EMPTY_ && !rg.node.HadPreviousLeader() {
resp.Error = jsNoConsumerErr
s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
}
}
return
}
}

7
server/jetstream_cluster_test.go
View File

@@ -5839,14 +5839,9 @@ func TestJetStreamClusterCreateConcurrentDurableConsumers(t *testing.T) {
defer c.shutdown()
// Client for API requests.
nc, _ := jsClientConnect(t, c.randomServer())
nc, js := jsClientConnect(t, c.randomServer())
defer nc.Close()
js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
// Create origin stream, muct be R > 1
if _, err := js.AddStream(&nats.StreamConfig{Name: "ORDERS", Replicas: 3}); err != nil {
t.Fatalf("Unexpected error: %v", err)