anchor.diff:    {ok, State}.
   
   %%%_* Internal Functions =======================================================
  +maybe_get_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
  +    {ok, Offset} ->
  +      {ok, Offset};
  +    {error, no_available_offsets} ->
  +      %% use new offset in this case and let the client handle the issue later
  +      {ok, NewOffset}
  +  end.
   
   get_valid_offset(SocketPid, InitialOffset, Topic, Partition) ->
     Request = #offset_request{ topic = Topic
  
positive.code:           {noreply, State};
          {shutdown, {ok, CbState, NewCbOptions}} ->
            {ok, State1} = update_cb_options(NewCbOptions, State0),
            State = State1#state{ cb_pending = CbPending
                                , cb_pending_cnt = CbPendingCnt
                                , cb_state = CbState},
            {noreply, State};
          Other ->
            {stop, {callback_error, Other}, State0}
  
negative.code:           {noreply, State0#state{cb_state = NewCbState}};
          {ok, NewCbState, NewCbOptions} ->
            {ok, State} = update_cb_options(NewCbOptions, State0),
            OldOffset = State0#state.offset,
            NewOffset0 = State#state.offset,
            SocketPid = State#state.socket_pid,
            {ok, NewOffset} = maybe_get_valid_offset(SocketPid, OldOffset,
                                                     NewOffset0, Topic, Partition),
            erlang:send_after(?ERROR_COOLDOWN, self(), ?SEND_FETCH_REQUEST),
            {noreply, State#state{cb_state = NewCbState, offset = NewOffset}};
          stop ->
            {stop, Error, State0};
          Other ->
  