anchor.diff:        {stop, {broker_error, Error}, State0};
       {ok, State} ->
         MessageSet = brod_utils:fetch_response_to_message_set(R),
  -      exec_callback(State#state.callback, State#state.callback3, MessageSet),
  +      exec_callback(State#state.callback, MessageSet),
         ok = send_fetch_request(State),
         {noreply, State};
       {empty, State} ->
  
positive.code:     false -> false
    end.
  
  exec_callback(Callback, undefined, MessageSet) ->
    Callback(MessageSet);
  exec_callback(undefined, Callback3, #message_set{messages = Messages}) ->
    F = fun(#message{key = K, value = V, offset = Offset}) ->
            Callback3(K, V, Offset)
        end,
    lists:foreach(F, Messages).
  
  do_debug(Pid, Debug) ->
    {ok, #socket{pid = Sock}} = get_socket(Pid),
  
negative.code:                , socket        :: #socket{}
                 , topic         :: binary()
                 , partition     :: integer()
                 , callback      :: callback_fun()
                 , offset        :: integer()
                 , max_wait_time :: integer()
                 , min_bytes     :: integer()
  