anchor.diff:              #state{socket_pid = Pid} = State) ->
     {stop, {socket_down, Reason}, State};
   %% callback completed
  -handle_info({handle_messages_result, Ref, Result}, State) ->
  -  [Ref | Rest] = State#state.cb_pending, %% assert
  -  CbPendingCnt = State#state.cb_pending_cnt - 1,
  -  NewState = State#state{ cb_pending     = Rest
  -                        , cb_pending_cnt = CbPendingCnt
  -                        },
  -  case Result of
  -    {ok, HandleMessagesResult} ->
  -      do_handle_messages_result(HandleMessagesResult, NewState);
  +handle_info({handle_messages_result, Ref, Result}, State0) ->
  +  [Ref | Rest] = State0#state.cb_pending, %% assert
  +  CbPendingCnt = State0#state.cb_pending_cnt - 1,
  +  State1 = State0#state{ cb_pending     = Rest
  +                       , cb_pending_cnt = CbPendingCnt
  +                       },
  +  case do_handle_messages_result(Result, State1) of
  +    {ok, State} ->
  +      maybe_handle_last_error(State);
       {error, Reason} ->
  -      {stop, {callback_crash, Reason}, NewState}
  +      {stop, Reason, State0}
     end;
   handle_info(Info, State) ->
     error_logger:warning_msg("~p [~p] ~p got unexpected info: ~p",
  
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
  