anchor.diff:      ?rebalance_warning("Unhandled cast: Msg = ~p, State = ~p", [Msg, State]),
       {noreply, State, ParentState}.
   
  -process_add_close_stream_response(Header, PendingPartitions, Errors) ->
  +process_stream_response(Header, PendingPartitions, Errors, SuccessPred) ->
       case lists:keytake(Header#mc_header.opaque, 1, PendingPartitions) of
           {value, {Partition} , N} ->
  -            case Header#mc_header.status of
  -                ?SUCCESS ->
  +            Status = Header#mc_header.status,
  +            case SuccessPred(Status) of
  +                true ->
                       {Partition, N, Errors};
  -                Status ->
  +                false ->
                       {error, N, [{Status, Partition} | Errors]}
               end;
           false ->
  
positive.code:             ParentState) ->
      {Header, _Body} = mc_binary:decode_packet(Packet),
  
      {Partition, NewToClose, NewErrors} = process_add_close_stream_response(Header, ToClose, Errors),
      NewStreamState = StreamState#stream_state{to_close_on_producer = NewToClose,
                                                errors = NewErrors},
  
  
negative.code:               = State, ParentState) ->
      {Header, _Body} = mc_binary:decode_packet(Packet),
  
      {Partition, NewToClose, NewErrors} = process_close_stream_response(Header, ToClose, Errors),
      NewStreamState = StreamState#stream_state{to_close = NewToClose, errors = NewErrors},
  
      NewState =
  