anchor.diff:    {ok, State}.
   
   %%%_* Internal Functions =======================================================
  +
  +do_handle_messages_result({ok, CbState}, State0) ->
  +  State1 = State0#state{cb_state = CbState},
  +  State2 = maybe_spawn_handle_messages(State1),
  +  {ok, State} = maybe_send_fetch_request(State2),
  +  {noreply, State};
  +do_handle_messages_result({ok, CbState, CbOptions}, State0) ->
  +  {ok, State1} = update_cb_options(CbOptions, State0),
  +  OldOffset = State0#state.offset,
  +  NewOffset0 = State1#state.offset,
  +  SocketPid = State1#state.socket_pid,
  +  Topic = State1#state.topic,
  +  Partition = State1#state.partition,
  +  {ok, NewOffset} = maybe_fetch_valid_offset(SocketPid, OldOffset,
  +                                             NewOffset0, Topic,
  +                                             Partition),
  +  State2 = State1#state{ cb_state = CbState
  +                       , offset   = NewOffset
  +                       },
  +  State3 = maybe_spawn_handle_messages(State2),
  +  {ok, State} = maybe_send_fetch_request(State3),
  +  {noreply, State};
  +do_handle_messages_result({error, Reason}, State) ->
  +  {stop, {callback_error, Reason}, State}.
  +
  +maybe_spawn_handle_messages(#state{ cb_pending = [F | Rest]
  +                                  , cb_state = CbState
  +                                  } = State) when is_function(F) ->
  +  Ref = make_ref(),
  +  Parent = self(),
  +  spawn_link(
  +    fun() ->
  +      Result =
  +        try
  +          {ok, F(CbState)}
  +        catch C : E ->
  +          {error, {C, E, erlang:get_stacktrace()}}
  +        end,
  +      Parent ! {handle_messages_result, Ref, Result}
  +    end),
  +  State#state{cb_pending = [Ref | Rest]};
  +maybe_spawn_handle_messages(State) ->
  +  State.
  +
   maybe_fetch_valid_offset(_SocketPid, Offset, Offset, _Topic, _Partition) ->
     {ok, Offset};
   maybe_fetch_valid_offset(SocketPid, _OldOffset, NewOffset, Topic, Partition) ->
  
positive.code:                                       State#state.prefetch_count),
    Offset = proplists:get_value(offset, NewCbOptions,
                                 State#state.offset),
    {ok, State#state{ offset = Offset
                    , min_bytes = MinBytes
                    , max_bytes = MaxBytes
                    , max_wait_time = MaxWaitTime
                    , sleep_timeout = SleepTimeout
                    , prefetch_count = PrefetchCount}}.
  
  %%%_* Tests ====================================================================
  
  
negative.code:             #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);
      {error, Reason} ->
        {stop, {callback_crash, Reason}, NewState}
    end;
  handle_info(Info, State) ->
    error_logger:warning_msg("~p [~p] ~p got unexpected info: ~p",
  