Original payload lost on reqeue

[utopos]

Processing failure requeues the message with the last updated payload:

  @impl Broadway.Acknowledger
  def ack(ack_ref, _successful, failed) do
    ack_options = :persistent_term.get(ack_ref)
    requeue? = ack_options[:on_failure] == :requeue

    requeue =
      failed
      |> Enum.filter(&ack?(&1, requeue?))
      |> Enum.map(& &1.data)

    Buffer.push(ack_options.buffer, requeue)

    :ok
  end

Expected behavior would to be requeue with initial payload.

[elliotekj]

Thanks for reporting! I don't have much spare time at the moment but I should be able to get a fix in this week.

[utopos]

Hey, I wrote a quick fix to it. Please have a look: 7e4cc89