anchor.diff:                = State, ParentState) ->
       {Header, _Body} = mc_binary:decode_packet(Packet),
   
  -    {Partition, NewToClose, NewErrors} = process_add_close_stream_response(Header, ToClose, Errors),
  +    {Partition, NewToClose, NewErrors} = process_close_stream_response(Header, ToClose, Errors),
       NewStreamState = StreamState#stream_state{to_close = NewToClose, errors = NewErrors},
   
       NewState =
  
positive.code:               = State, ParentState) ->
      {Header, Body} = mc_binary:decode_packet(Packet),
  
      {Partition, NewToAdd, NewErrors} = process_add_close_stream_response(Header, ToAdd, Errors),
      NewStreamState = StreamState#stream_state{to_add = NewToAdd, errors = NewErrors},
  
      NewState =
  
negative.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 ->
                      {error, N, [{Status, Partition} | Errors]}
              end;
          false ->
  