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:   Offset1 = proplists:get_value(offset, CbOptions, Offset0),
  
    %% 3. Retrieve valid starting offset
    case get_valid_offset(SocketPid, Offset1, Topic, Partition) of
      {error, Error} ->
        {stop, {error, Error}, State0};
      {ok, Offset} ->
  
negative.code:           SocketPid = State1#state.socket_pid,
            Topic = State1#state.topic,
            Partition = State1#state.partition,
            {ok, NewOffset} = maybe_fetch_valid_offset(SocketPid, OldOffset,
                                                       NewOffset0, Topic,
                                                       Partition),
            State1 = State1#state{ cb_pending = CbPending
                                 , cb_pending_cnt = CbPendingCnt
                                 , cb_state = CbState
  