anchor.diff:                       , last_offset = LastOffset
                        , messages = Messages} = PM,
     CbMod = State0#state.cb_mod,
  -  CbState = State0#state.cb_state,
     case brod_kafka:is_error(ErrorCode) of
       true ->
  +      Now = os:timestamp(),
         Error = {ErrorCode, brod_kafka_errors:desc(ErrorCode)},
  -      case CbMod:handle_error(Topic, Partition, Error, CbState) of
  -        {ok, NewCbState} ->
  -          erlang:send_after(?ERROR_COOLDOWN, self(), ?SEND_FETCH_REQUEST),
  -          {noreply, State0#state{cb_state = NewCbState}};
  -        {ok, NewCbState, NewCbOptions} ->
  -          {ok, State} = update_cb_options(NewCbOptions, State0),
  -          OldOffset = State0#state.offset,
  -          NewOffset0 = State#state.offset,
  -          SocketPid = State#state.socket_pid,
  -          {ok, NewOffset} = maybe_fetch_valid_offset(SocketPid, OldOffset,
  -                                                     NewOffset0, Topic, Partition),
  -          erlang:send_after(?ERROR_COOLDOWN, self(), ?SEND_FETCH_REQUEST),
  -          {noreply, State#state{cb_state = NewCbState, offset = NewOffset}};
  -        stop ->
  -          {stop, Error, State0};
  -        Other ->
  -          {stop, {callback_error, Other, Error}, State0}
  -      end;
  +      F = fun(CbStateIn) ->
  +            CbMod:handle_error(Topic, Partition, Error, CbStateIn)
  +          end,
  +      State = State0#state{last_error = ?KAFKA_ERROR(Now, ErrorCode, F)},
  +      maybe_handle_last_error(State);
       false ->
  -      SleepTimeout = State0#state.sleep_timeout,
         case Messages of
  -        [] when SleepTimeout =:= 0 ->
  -          {ok, NewCorrId} = send_fetch_request(State0),
  -          {noreply, State0#state{corr_id = NewCorrId}};
  -        [] when SleepTimeout > 0 ->
  -          erlang:send_after(SleepTimeout, self(), ?SEND_FETCH_REQUEST),
  +        [] ->
  +          maybe_delay_fetch_request(State0#state.sleep_timeout),
             {noreply, State0};
           [_|_] ->
             F = fun(CbStateIn) ->
  
positive.code: -define(DEFAULT_MAX_WAIT_TIME, 1000). % 1 sec
  -define(DEFAULT_SLEEP_TIMEOUT, 1000). % 1 sec
  -define(DEFAULT_PREFETCH_COUNT, 1).
  
  -define(SEND_FETCH_REQUEST, send_fetch_request).
  -define(ERROR_COOLDOWN, 1000).
  
  %%%_* APIs =====================================================================
  %% @equiv start_link(ClientId, Topic, Partition, Config, [])
  
negative.code:                        , offset   = NewOffset
                         },
    State3 = maybe_spawn_handle_messages(State2),
    maybe_send_fetch_request(State3);
  do_handle_messages_result({error, Reason}, _State) ->
    {error, {callback_error, Reason}}.
  
  maybe_spawn_handle_messages(#state{ cb_pending = [F | Rest]
                                    , cb_state = CbState
  