[4.3.1+] Consumers do not reprocess message if callback function returned unexpected response

[L4vlet]

Greetings.

We have a simple test that validates that crashes inside the message handler module leave messages uncommitted and available for reprocessing.

test "messages are reprocessed, if anything fails" do
    topic = "manual_topic"
    {:ok, agent_pid} = start_supervised({Agent, fn -> 0 end})

    {:ok, _, producer_client_id} =
      TestBrod.start_consumer_producer(topic,
        consumer: [
          handler: ConsumerHandlerError,
          extra: [test_agent_pid: agent_pid],
          group_id: group_id = "manual_tests",
          begin_offset: :latest
        ]
      )

    payload = Utils.random_str()
    key = Utils.random_str()

    assert :ok = Producer.publish(producer_client_id, topic, key, payload)

    assert_receive {:processed_error, ^key, ^payload}, @consumer_timeout
    assert_receive {:processed, ^key, ^payload}, @consumer_timeout
end

ConsumerHandlerError is a simple handle that uses Agent to simulate transient errors. Something like:

def handle_message(_message, _meta, %{extra: opts}) do
    case Keyword.fetch(opts, :test_agent_pid) do
      {:ok, agent_pid} ->
        invocation = Agent.get_and_update(agent_pid, fn state -> {state, state + 1} end)

        if invocation == 0 do
          {:error, "something unexpected happened"}
        else
          :ok
        end

      :error ->
        :error
    end
end

Internally it is wrapped to send :processed_* messages to the test PID.

Starting from version 4.3.1 it no longer works.
Failed messages are not reprocessed despite all consumers successfully restarted by the Supervisor after receiving an unexpected result.

If you set begin_offset: :earliest it passes as before.

I investigated changes related to the 4.3.1 version (one, two) but could not link them to the observed behaviour.
Am I wrong to believe this should not happen?

Test consumer configuration:

[
    handler: ConsumerHandlerError,
    auto_start: true,
    topics: ["manual_topic"],
    min_bytes: 0,
    max_wait_time: 0,
    message_type: :message_set,
    group_id: "manual_tests",
    begin_offset: :latest
  ]

[mikpe]

I'd suggest you use git bisect to identify the cause of the breakage.