anchor.diff:    {ok, _} = gen:call(Pid, system, {debug, Debug}, infinity),
     ok.
   
  -update_cb_options(NewCbOptions, State) ->
  +update_cb_options(NewCbOptions, #state{offset = OldOffset} = State) ->
     MinBytes = proplists:get_value(min_bytes,NewCbOptions,
                                    State#state.min_bytes),
     MaxBytes = proplists:get_value(max_bytes, NewCbOptions,
  
positive.code:   {ok, State}.
  
  %%%_* Internal Functions =======================================================
  maybe_fetch_valid_offset(_SocketPid, Offset, Offset, _Topic, _Partition) ->
    {ok, Offset};
  maybe_fetch_valid_offset(SocketPid, _OldOffset, NewOffset, Topic, Partition) ->
  
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",
  