anchor.diff:      %%   notification before handling our close stream request
       %%
       %% Because of this we ignore ?KEY_ENOENT errors.
  -    process_stream_response(Header, PendingPartitions, Errors,
  -                            fun allow_success_enoent/1).
  +    Status =:= ?SUCCESS orelse Status =:= ?KEY_ENOENT.
  +
  +process_add_stream_response(Header, PendingPartitions, Errors) ->
  +    process_stream_response(Header, PendingPartitions, Errors, add_stream).
  +
  +process_close_stream_response(Header, PendingPartitions, Errors) ->
  +    process_stream_response(Header, PendingPartitions, Errors, close_stream).
   
   maybe_reply_setup_streams(#state{state = StreamState} = State) ->
       case {StreamState#stream_state.to_add, StreamState#stream_state.to_close,
  
positive.code:     ?rebalance_warning("Unhandled cast: Msg = ~p, State = ~p", [Msg, State]),
      {noreply, State, ParentState}.
  
  process_stream_response(Header, PendingPartitions, Errors, SuccessPred) ->
      case lists:keytake(Header#mc_header.opaque, 1, PendingPartitions) of
          {value, {Partition} , N} ->
              Status = Header#mc_header.status,
              case SuccessPred(Status) of
                  true ->
                      {Partition, N, Errors};
                  false ->
  
negative.code:             erlang:error({unrecognized_opaque, Header#mc_header.opaque, PendingPartitions})
      end.
  
  is_successful_stream_response(add_stream, Status) ->
      Status =:= ?SUCCESS;
  is_successful_stream_response(close_stream, Status) ->
      %% It's possible that the stream is already closed in the following cases:
      %%
      %%   - on the producer and consumer sides, if the vbucket has been moved
  