anchor.diff:      space_id = SpaceId,
       stream = Stream
   }, Docs = [Doc | RestDocs]) ->
  -    try
  -        State2 = consume_change(State, Doc),
  -        consume_change_and_enter_retry_mode_on_error(State2, RestDocs)
  +    State2 = try
  +        consume_change(State, Doc)
       catch
           Error:Reason ->
               ?error_stacktrace("Unexpected error ~p:~p in harvest_stream ~p "
               "handling changes for harvester ~p in space ~p",
                   [Error, Reason, Id, HarvesterId, SpaceId]),
  -            catch gen_server:stop(Stream),
  +            stop_changes_stream(Stream),
               enter_retry_mode(Docs, State)
  +    end,
  +    case State2#state.mode of
  +        retry ->
  +            State2;
  +        _ ->
  +            consume_change_and_enter_retry_mode_on_error(State2, RestDocs)
       end;
   consume_change_and_enter_retry_mode_on_error(State, Doc = #document{}) ->
       consume_change_and_enter_retry_mode_on_error(State, [Doc]).
  
positive.code:     {noreply, NewState :: state()} |
      {noreply, NewState :: state(), timeout() | hibernate} |
      {stop, Reason :: term(), NewState :: state()}.
  handle_cast(_Request, #state{mode = Mode} = State)
      when Mode =:= retry orelse Mode =:= resuming ->
      % ignore because stream is in retry mode
      {noreply, State};
  handle_cast({change, {ok, end_of_stream}}, State = #state{mode = retry}) ->
      % all changes sent by changes_stream before stopping it have already arrived
      {noreply, State#state{changes_flushed = true}};
  
negative.code:         [{since, Since}, {until, infinity}], []
      ).
  
  -spec stop_changes_stream(pid()) -> ok.
  stop_changes_stream(Stream) ->
      try
          gen_server:stop(Stream)
      catch
          exit:noproc ->
              ok
      end.
  
  %%-------------------------------------------------------------------
  %% @private
  %% @doc
  