Remove ordered consumer on reset and fix recreated consumer name

js.go

@@ -1241,6 +1241,10 @@ func (sub *Subscription) deleteConsumer() error {
 		sub.mu.Unlock()
 		return nil
 	}
+	if jsi.stream == _EMPTY_ || jsi.consumer == _EMPTY_ {
+		sub.mu.Unlock()
+		return nil
+	}
 	stream, consumer := jsi.stream, jsi.consumer
 	js := jsi.js
 	sub.mu.Unlock()
@@ -2047,7 +2051,26 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		js := jsi.js
 		sub.mu.Unlock()
 
-		consName := nuid.Next()
+		// Attempt to delete the existing consumer.
+		// If ErrConsumerNotFound is returned, it means that the consumer was already deleted
+		// by the server, so we can proceed to create a new consumer.
+		err := sub.deleteConsumer()
+		if err != nil && !errors.Is(err, ErrConsumerNotFound) {
+			var apiErr *APIError
+			if errors.Is(err, ErrJetStreamNotEnabled) || errors.Is(err, ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
+				// if creating consumer failed, retry
+				return
+			} else if errors.As(err, &apiErr) && apiErr.ErrorCode == JSErrCodeInsufficientResourcesErr {
+				// retry for insufficient resources, as it may mean that client is connected to a running
+				// server in cluster while the server hosting R1 JetStream resources is restarting
+				return
+			}
+			pushErr(err)
+		}
+		sub.mu.Lock()
+		jsi.consumer = ""
+		sub.mu.Unlock()
+		consName := getHash(nuid.Next())
 		cinfo, err := js.upsertConsumer(jsi.stream, consName, cfg)
 		if err != nil {
 			var apiErr *APIError

test/js_test.go

@@ -5807,7 +5807,6 @@ func TestJetStreamSubscribe_ReplayPolicy(t *testing.T) {
 	// Enough time to get the next message according to the original playback.
 	_, err = sub.NextMsg(110 * time.Millisecond)
 	if err != nil {
-
 		t.Fatalf("Unexpected error: %v", err)
 	}
 }
@@ -9382,17 +9381,32 @@ func TestJetStreamOrderedConsumerRecreateAfterReconnect(t *testing.T) {
 	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(time.Second)
+	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)
@@ -9403,13 +9417,34 @@ func TestJetStreamOrderedConsumerRecreateAfterReconnect(t *testing.T) {
 	case <-time.After(10 * time.Second):
 		t.Fatalf("Did not receive consumer not active error")
 	}
+	consDeleteMsg, err := apiSub.NextMsg(2 * time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !strings.HasPrefix(consDeleteMsg.Subject, "$JS.API.CONSUMER.DELETE.") {
+		t.Fatalf("Unexpected message subject: %q", consDeleteMsg.Subject)
+	}
+	consCreateMsg, err := apiSub.NextMsg(2 * time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if !strings.HasPrefix(consCreateMsg.Subject, "$JS.API.CONSUMER.CREATE.") {
+		t.Fatalf("Unexpected message subject: %q", consCreateMsg.Subject)
+	}
 	if _, err := js.Publish("FOO.A", []byte("msg 2")); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	msg, err = sub.NextMsg(time.Second)
+	msg, err = sub.NextMsg(2 * time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	consInfo, err = sub.ConsumerInfo()
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+	if consInfo.Name == consName || len(consInfo.Name) != 8 {
+		t.Fatalf("Unexpected consumer name: %q", consInfo.Name)
+	}
 
 	// make sure we pick up where we left off
 	if string(msg.Data) != "msg 2" {