Question: Handle Leadership Change in PullConsumers

[ludusrusso]

Hi guys, I'm running a nats jetstream cluster on kubernetes with the official helm chart and 3 replicas.
In some cases I notice an error of Leadership Change in pull consumer, here is an example code:

	ctx, cancel := context.WithTimeout(mainCtx, 10*time.Second)
	defer cancel()
	msgs, err := sub.Fetch(3, nats.Context(ctx))
	if err != nil {
		if errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
			return nil
		}
		return fmt.Errorf("cannot fetch: %w", err)
	}
	
	// handle msgs

Hi was wondering, which is the best way to handle this errors? Should I recreate the subscriber or I can just retry to Fetch?

[wallyqs]

Short answer I think would be to retry, but also 'Leadership Change' is a 409 status sent from the server and a temporary error that I think the client ought to handle internally similar to how 408 is being handled until the deadline expires:

nats.go/nats.go

Lines 3319 to 3320 in 1714729

	reqTimeoutSts = "408"
	controlMsg = "100"

nats.go/js.go

Lines 2450 to 2455 in 1714729

	case reqTimeoutSts:
	// Older servers may send a 408 when a request in the server was expired
	// and interest is still found, which will be the case for our
	// implementation. Regardless, ignore 408 errors until receiving at least
	// one message.
	err = ErrTimeout

[ludusrusso]

Ok, so for now I can workaround it with this

if err != nil {
		if errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
			return nil
		} else if err.Error() == "nats: Leadership Change" {
			logrus.Errorf("nats leadership change, restarting loop: %v", err)
			return nil
		}
		return fmt.Errorf("cannot fetch: %w", err)
	}

Maybe I can push a PR to handle this bug