Function TestJetStreamResetConsumer

jetstream/test/consumer_test.go:1603–1718

(t *testing.T)

}

func TestJetStreamResetConsumer(t *testing.T) {
	srv := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, srv)

	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	ctx := context.Background()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	for i := range 10 {
		if _, err := js.Publish(ctx, "foo", fmt.Appendf(nil, "msg-%d", i)); err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
	}

	cons, err := js.CreateConsumer(ctx, "TEST", jetstream.ConsumerConfig{
		Durable:   "cons",
		AckPolicy: jetstream.AckExplicitPolicy,
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	t.Run("reset to ack floor", func(t *testing.T) {
		// Drain 5 messages without ack so ack_floor stays 0 and the
		// next fetch after reset is observably different from no-op.
		batch, err := cons.Fetch(5)
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		for range batch.Messages() {
		}

		resp, err := js.ResetConsumer(ctx, "TEST", "cons")
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		if resp.ResetSeq != 1 {
			t.Fatalf("Expected ResetSeq=1, got %d", resp.ResetSeq)
		}

		next, err := cons.Fetch(1)
		if err != nil {