mirror of
https://github.com/nats-io/nats.go.git
synced 2026-04-22 15:57:08 +08:00
[IMPROVED] Return more specific cons info error on ordered consumer recreation (#1931)
Signed-off-by: Piotr Piotrowski <piotr@synadia.com>
This commit is contained in:
@@ -2836,6 +2836,10 @@ func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
|
||||
sub.mu.Lock()
|
||||
// TODO(dlc) - Better way to mark especially if we attach.
|
||||
if sub.jsi == nil || sub.jsi.consumer == _EMPTY_ {
|
||||
if sub.jsi.ordered {
|
||||
sub.mu.Unlock()
|
||||
return nil, ErrConsumerInfoOnOrderedReset
|
||||
}
|
||||
sub.mu.Unlock()
|
||||
return nil, ErrTypeSubscription
|
||||
}
|
||||
|
||||
@@ -150,6 +150,9 @@ var (
|
||||
// ErrConsumerLeadershipChanged is returned when pending requests are no longer valid after leadership has changed
|
||||
ErrConsumerLeadershipChanged JetStreamError = &jsError{message: "Leadership Changed"}
|
||||
|
||||
// ErrConsumerInfoOnOrderedReset is returned when attempting to fetch consumer info for an ordered consumer that is currently being recreated.
|
||||
ErrConsumerInfoOnOrderedReset JetStreamError = &jsError{message: "cannot fetch consumer info; ordered consumer is being reset"}
|
||||
|
||||
// ErrNoHeartbeat is returned when no heartbeat is received from server when sending requests with pull consumer.
|
||||
ErrNoHeartbeat JetStreamError = &jsError{message: "no heartbeat received"}
|
||||
|
||||
|
||||
+15
-3
@@ -10105,9 +10105,21 @@ 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)
|
||||
var infoErr error
|
||||
for range 5 {
|
||||
consInfo, infoErr = sub.ConsumerInfo()
|
||||
if infoErr != nil {
|
||||
if errors.Is(infoErr, nats.ErrConsumerInfoOnOrderedReset) {
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
continue
|
||||
}
|
||||
t.Fatalf("Unexpected error: %v", err)
|
||||
}
|
||||
infoErr = nil
|
||||
break
|
||||
}
|
||||
if infoErr != nil {
|
||||
t.Fatalf("Unexpected error: %v", infoErr)
|
||||
}
|
||||
if consInfo.Name == consName || len(consInfo.Name) != 8 {
|
||||
t.Fatalf("Unexpected consumer name: %q", consInfo.Name)
|
||||
|
||||
Reference in New Issue
Block a user