anchor.diff:      {error, Error} ->
         {reply, {error, Error}, State0};
       {ok, Offset} ->
  -      State1 = case erlang:fun_info(Callback, arity) of
  -                 {arity, 1} ->
  -                   State0#state{callback = Callback, callback3 = undefined};
  -                 {arity, 3} ->
  -                   State0#state{callback = undefined, callback3 = Callback}
  -               end,
  -      State = State1#state{ offset = Offset
  +      State = State0#state{ callback      = Callback
  +                          , offset        = Offset
                             , max_wait_time = MaxWaitTime
  -                          , min_bytes = MinBytes
  -                          , max_bytes = MaxBytes},
  +                          , min_bytes     = MinBytes
  +                          , max_bytes     = MaxBytes},
         ok = send_fetch_request(State),
         {reply, ok, State}
     end;
  
positive.code:                , socket        :: #socket{}
                 , topic         :: binary()
                 , partition     :: integer()
                 , callback      :: fun((#message_set{}) -> ok)
                 , callback3     :: fun((binary(), binary(), integer()) -> ok)
                 , offset        :: integer()
                 , max_wait_time :: integer()
                 , min_bytes     :: integer()
  
negative.code:     false -> false
    end.
  
  exec_callback(Callback, MessageSet) ->
    case erlang:fun_info(Callback, arity) of
      {arity, 1} ->
        Callback(MessageSet);
      {arity, 3} ->
        F = fun(#message{key = K, value = V, offset = Offset}) ->
                Callback(K, V, Offset)
            end,
        lists:foreach(F, MessageSet#message_set.messages)
    end.
  
  do_debug(Pid, Debug) ->
    {ok, #socket{pid = Sock}} = get_socket(Pid),
  