fix: throw on retry when payload is consume by downstream

fix: throw on retry when payload is consume by downstream #3463