Merge pull request #1151 from sata-form3/sata-create-consumer-leaders…

…hip-changed-error

Provide a way to handle 409 Leadership Changed errors in client

js.go

@@ -2580,6 +2580,11 @@ func checkMsg(msg *Msg, checkSts, isNoWait bool) (usrMsg bool, err error) {
 			err = ErrConsumerDeleted
 			break
 		}
+
+		if strings.Contains(strings.ToLower(string(msg.Header.Get(descrHdr))), "leadership change") {
+			err = ErrConsumerLeadershipChanged
+			break
+		}
 		fallthrough
 	default:
 		err = fmt.Errorf("nats: %s", msg.Header.Get(descrHdr))

jserrors.go

@@ -101,6 +101,9 @@ var (
 	// ErrConsumerDeleted is returned when attempting to send pull request to a consumer which does not exist
 	ErrConsumerDeleted JetStreamError = &jsError{message: "consumer deleted"}
 
+	// ErrConsumerLeadershipChanged is returned when pending requests are no longer valid after leadership has changed
+	ErrConsumerLeadershipChanged JetStreamError = &jsError{message: "Leadership Changed"}
+
 	// DEPRECATED: ErrInvalidDurableName is no longer returned and will be removed in future releases.
 	// Use ErrInvalidConsumerName instead.
 	ErrInvalidDurableName = errors.New("nats: invalid durable name")