Repository: nats-io/nats.go (go)

Issue:
Restarting Ordered Consumer by Start Time
https://github.com/nats-io/nats.go/blob/7c684685694621ba5e98b2471b3a11618d03da6f/js.go#L1983

If we are using an ordered consumer with a particular start time. On the consumer restarting I think it tries to resume at the sequence it last left off at, but this is in conflict with the start time. If the consumer was already successfully running with the deliver policy `DeliverByStartTime` then it should clear the start time option when resuming with `DeliverByStartSequence`

Pull request description:
Unset start time for ordered consumer on retry
If an ordered consumer was configured with a start time, on retry, the copied consumer config would start with sequence but the existing start time opt would conflict causing a config error.

Fix #1281

---

Just noting that I added the `StartTime()` opt to the test which, without clearing start time on retry (the fix), it will demonstrate the behavior and get this message:

```
nats: consumer delivery policy is deliver by start sequence, but optional start time is also set: recreating ordered consumer on connection [4] for subscription on "a"
```

Patch:
diff --git a/js.go b/js.go
index 9ecf1f5..3559f5a 100644
--- a/js.go
+++ b/js.go
@@ -1991,6 +1991,9 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		cfg.DeliverSubject = newDeliver
 		cfg.DeliverPolicy = DeliverByStartSequencePolicy
 		cfg.OptStartSeq = sseq
+		// In case the consumer was created with a start time, we need to clear it
+		// since we are now using a start sequence.
+		cfg.OptStartTime = nil
 
 		js := jsi.js
 		sub.mu.Unlock()


Test patch:
diff --git a/js_test.go b/js_test.go
index c73a261..f01547e 100644
--- a/js_test.go
+++ b/js_test.go
@@ -114,6 +114,9 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	// Will be used as start time to validate proper reset to sequence on retries.
+	startTime := time.Now()
+
 	// Create a sample asset.
 	msg := make([]byte, 1024*1024)
 	rand.Read(msg)
@@ -186,7 +189,7 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 			rmsg = append(rmsg, m.Data...)
 		}
 		// OrderedConsumer does not need HB, it sets it on its own, but for test we override which is ok.
-		sub, err := js.Subscribe("a", cb, OrderedConsumer(), IdleHeartbeat(250*time.Millisecond))
+		sub, err := js.Subscribe("a", cb, OrderedConsumer(), IdleHeartbeat(250*time.Millisecond), StartTime(startTime))
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -208,7 +211,7 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 		var rmsg []byte
 
 		// OrderedConsumer does not need HB, it sets it on its own, but for test we override which is ok.
-		sub, err := js.SubscribeSync("a", OrderedConsumer(), IdleHeartbeat(250*time.Millisecond))
+		sub, err := js.SubscribeSync("a", OrderedConsumer(), IdleHeartbeat(250*time.Millisecond), StartTime(startTime))
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}