anchor.diff:    {ok, State}.
   
   %%%_* Internal Functions =======================================================
  -maybe_get_valid_offset(_SocketPid, Offset, Offset, _Topic, _Partition) ->
  +maybe_fetch_valid_offset(_SocketPid, Offset, Offset, _Topic, _Partition) ->
     {ok, Offset};
  -maybe_get_valid_offset(SocketPid, _OldOffset, NewOffset, Topic, Partition) ->
  -  case get_valid_offset(SocketPid, NewOffset, Topic, Partition) of
  +maybe_fetch_valid_offset(SocketPid, _OldOffset, NewOffset, Topic, Partition) ->
  +  case fetch_valid_offset(SocketPid, NewOffset, Topic, Partition) of
       {ok, Offset} ->
         {ok, Offset};
       {error, no_available_offsets} ->
  
positive.code:           SocketPid = State1#state.socket_pid,
            Topic = State1#state.topic,
            Partition = State1#state.partition,
            {ok, NewOffset} = maybe_get_valid_offset(SocketPid, OldOffset,
                                                     NewOffset0, Topic,
                                                     Partition),
            State1 = State1#state{ cb_pending = CbPending
                                 , cb_pending_cnt = CbPendingCnt
                                 , cb_state = CbState
  
negative.code:   Offset1 = proplists:get_value(offset, CbOptions, Offset0),
  
    %% 3. Retrieve valid starting offset
    case fetch_valid_offset(SocketPid, Offset1, Topic, Partition) of
      {error, Error} ->
        {stop, {error, Error}, State0};
      {ok, Offset} ->
  