diff --git a/server/consumer.go b/server/consumer.go index b3b99eb092..73bd2e428e 100644 --- a/server/consumer.go +++ b/server/consumer.go @@ -1760,7 +1760,7 @@ func (o *consumer) readStoredState() error { return nil } state, err := o.store.State() - if err == nil && state != nil { + if err == nil && state != nil && state.Delivered.Consumer != 0 { o.applyState(state) if len(o.rdc) > 0 { o.checkRedelivered() diff --git a/server/filestore.go b/server/filestore.go index 2c37c43f83..28ac847f69 100644 --- a/server/filestore.go +++ b/server/filestore.go @@ -5509,11 +5509,10 @@ func (o *consumerFileStore) State() (*ConsumerState, error) { o.mu.Lock() defer o.mu.Unlock() - var state *ConsumerState + state := &ConsumerState{} // See if we have a running state or if we need to read in from disk. if o.state.Delivered.Consumer != 0 { - state = &ConsumerState{} state.Delivered = o.state.Delivered state.AckFloor = o.state.AckFloor if len(o.state.Pending) > 0 { diff --git a/server/filestore_test.go b/server/filestore_test.go index a631602430..a4a16da947 100644 --- a/server/filestore_test.go +++ b/server/filestore_test.go @@ -1892,7 +1892,7 @@ func TestFileStoreConsumer(t *testing.T) { if err != nil { t.Fatalf("Unexepected error: %v", err) } - if state, err := o.State(); state != nil || err != nil { + if state, err := o.State(); err != nil || state.Delivered.Consumer != 0 { t.Fatalf("Unexpected state or error: %v", err) } diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go index a6dc64170e..0e85c1e863 100644 --- a/server/jetstream_cluster.go +++ b/server/jetstream_cluster.go @@ -3106,11 +3106,6 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) { lastSnap = snap } } - } else { - // If we are here we may have no state but may be processing updates as a filtered consumer. - // Make sure the store does not grow too much, so similar to memory logic above, just compact. - _, _, applied := n.Progress() - n.Compact(applied) } } diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go index 4224d0fb23..af96996e3b 100644 --- a/server/jetstream_cluster_test.go +++ b/server/jetstream_cluster_test.go @@ -4309,7 +4309,7 @@ func TestJetStreamClusterStreamLeaderStepDown(t *testing.T) { t.Fatalf("Unexpected error: %v", err) } if cdResp.Error != nil { - t.Fatalf("Unexpected error: %+v", sdResp.Error) + t.Fatalf("Unexpected error: %+v", cdResp.Error) } checkFor(t, 2*time.Second, 50*time.Millisecond, func() error { @@ -10918,6 +10918,20 @@ func TestJetStreamClusterFilteredAndIdleConsumerNRGGrowth(t *testing.T) { if entries, _ := o.raftNode().Size(); entries > compactNumMin { t.Fatalf("Expected <= %d entries, got %d", compactNumMin, entries) } + + // Now make the consumer leader stepdown and make sure we have the proper snapshot. + resp, err := nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "dlc"), nil, time.Second) + require_NoError(t, err) + + var cdResp JSApiConsumerLeaderStepDownResponse + if err := json.Unmarshal(resp.Data, &cdResp); err != nil { + t.Fatalf("Unexpected error: %v", err) + } + if cdResp.Error != nil { + t.Fatalf("Unexpected error: %+v", cdResp.Error) + } + + c.waitOnConsumerLeader("$G", "TEST", "dlc") } func TestJetStreamClusterMirrorOrSourceNotActiveReporting(t *testing.T) {