move backoffMtx to cover update to backoffDuration

consumer.go

@@ -656,12 +656,13 @@ func (r *Consumer) startStopContinueBackoff(conn *Conn, signal backoffSignal) {
 	// prevent many async failures/successes from immediately resulting in
 	// max backoff/normal rate (by ensuring that we dont continually incr/decr
 	// the counter during a backoff period)
+	r.backoffMtx.Lock()
 	if r.inBackoffBlock() {
+		r.backoffMtx.Unlock()
 		return
 	}
 
 	// update backoff state
-	r.backoffMtx.Lock()
 	backoffUpdated := false
 	switch signal {
 	case resumeFlag:
@@ -677,7 +678,6 @@ func (r *Consumer) startStopContinueBackoff(conn *Conn, signal backoffSignal) {
 			backoffUpdated = true
 		}
 	}
-	r.backoffMtx.Unlock()
 
 	if r.backoffCounter == 0 && backoffUpdated {
 		// exit backoff
@@ -700,6 +700,7 @@ func (r *Consumer) startStopContinueBackoff(conn *Conn, signal backoffSignal) {
 			r.updateRDY(c, 0)
 		}
 	}
+	r.backoffMtx.Unlock()
 }
 
 func (r *Consumer) backoff() {