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,16 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		js := jsi.js
 		sub.mu.Unlock()
 
-		consName := nuid.Next()
+		sub.mu.Lock()
+		// Attempt to delete the existing consumer.
+		// We don't wait for the response since even if it's unsuccessful,
+		// inactivity threshold will kick in and delete it.
+		if jsi.consumer != _EMPTY_ {
+			go js.DeleteConsumer(jsi.stream, jsi.consumer)
+		}
+		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.") {
+		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.") {
+		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" {