Function TestJetStreamOrderedConsumerRecreateAfterReconnect

test/js_test.go:10051–10153

(t *testing.T)

}

func TestJetStreamOrderedConsumerRecreateAfterReconnect(t *testing.T) {
	s := RunBasicJetStreamServer()

	// monitor for ErrConsumerNotActive error and suppress logging
	hbMissed := make(chan struct{}, 10)
	errHandler := func(c *nats.Conn, s *nats.Subscription, err error) {
		if !errors.Is(err, nats.ErrConsumerNotActive) {
			t.Fatalf("Unexpected error: %v", err)
		}
		hbMissed <- struct{}{}
	}
	nc, js := jsClient(t, s, nats.ErrorHandler(errHandler))
	defer nc.Close()

	if _, err := js.AddStream(&nats.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}}); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	sub, err := js.SubscribeSync("FOO.A", nats.OrderedConsumer(), nats.IdleHeartbeat(100*time.Millisecond))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	consInfo, err := sub.ConsumerInfo()
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	consName := consInfo.Name
	// validate that the generated name of the consumer is 8
	// characters long (shorter than standard nuid)
	if len(consName) != 8 {
		t.Fatalf("Unexpected consumer name: %q", consName)
	}
	if _, err := js.Publish("FOO.A", []byte("msg 1")); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	msg, err := sub.NextMsg(2 * time.Second)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if string(msg.Data) != "msg 1" {
		t.Fatalf("Invalid msg value; want: 'msg 1'; got: %q", string(msg.Data))
	}

	apiSub, err := nc.SubscribeSync("$JS.API.CONSUMER.*.>")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	// restart the server
	s = restartBasicJSServer(t, s)
	defer shutdownJSServerAndRemoveStorage(t, s)

	// wait until we miss heartbeat
	select {
	case <-hbMissed:
	case <-time.After(10 * time.Second):
		t.Fatalf("Did not receive consumer not active error")
	}
	consDeleteMsg, err := apiSub.NextMsg(2 * time.Second)