anchor.diff:        {ok, NewOffset}
     end.
   
  -get_valid_offset(SocketPid, InitialOffset, Topic, Partition) ->
  +fetch_valid_offset(SocketPid, InitialOffset, Topic, Partition) ->
     Request = #offset_request{ topic = Topic
                              , partition = Partition
                              , time = InitialOffset
  
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:   {ok, State}.
  
  %%%_* Internal Functions =======================================================
  maybe_fetch_valid_offset(_SocketPid, Offset, Offset, _Topic, _Partition) ->
    {ok, Offset};
  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} ->
  