anchor.diff:            erlang:send_after(SleepTimeout, self(), ?SEND_FETCH_REQUEST),
             {noreply, State0};
           [_|_] ->
  -          CbMod = State0#state.cb_mod,
  -          CbState = State0#state.cb_state,
             F = fun() ->
                     Res = CbMod:handle_messages(Topic, Partition,HighWmOffset,
                                                 Messages, CbState),
  
positive.code:                          , CbState      :: any()) ->
    {ok, NewCbState :: any()} |
    {ok, NewCbState :: any(), NewCbOptions :: [{consumer_option(), any()}]}.
  
  -record(state, { cb_mod         :: atom()
                 , cb_state       :: any()
  
negative.code:                               , cb_state = CbState},
            {noreply, State};
          {shutdown, {ok, CbState, NewCbOptions}} ->
            {ok, State1} = update_cb_options(NewCbOptions, State0),
            State = State1#state{ cb_pending = CbPending
                                , cb_pending_cnt = CbPendingCnt
                                , cb_state = CbState},
            {noreply, State};
          Other ->
            {stop, {callback_error, Other}, State0}
  