anchor.diff:  %% PIPE Handle response from mapred_stream
   handle_info(#pipe_eoi{ref=ReqId},
               State=#state{req=#rpbmapredreq{},
  -                         req_ctx=#pipe_ctx{ref=ReqId}}) ->
  +                         req_ctx=#pipe_ctx{ref=ReqId,
  +                                           timer=Timer}}) ->
       NewState = send_msg(#rpbmapredresp{done = 1}, State),
  +    erlang:cancel_timer(Timer),
       {noreply, NewState#state{req = undefined, req_ctx = undefined}};
   
   handle_info(#pipe_result{ref=ReqId, from=PhaseId, result=Res},
  
positive.code: 
  -record(pipe_ctx, {pipe,     % pipe handling mapred request
                     ref,      % easier-access ref/reqid
                     sender}). % {pid(), monitor()} of process sending inputs
  
  -define(PROTO_MAJOR, 1).
  
negative.code:        true ->
              %% something went wrong sending inputs - tell the client
              %% about it, and shutdown the pipe
              erlang:cancel_timer(PipeCtx#pipe_ctx.timer),
              riak_pipe:destroy(PipeCtx#pipe_ctx.pipe),
              lager:error("Error sending inputs: ~p", [Reason]),
              NewState = send_error("~p", [bad_mapred_inputs], State),
              {noreply, NewState#state{req=undefined, req_ctx=undefined}}
      end;
  handle_info({pipe_timeout, Ref},
              State=#state{req=#rpbmapredreq{},
                           req_ctx=#pipe_ctx{ref=Ref,
                                             sender=Sender,
                                             pipe=Pipe}}) ->
      NewState = send_error("timeout", [], State),
      riak_pipe:destroy(Pipe),
      case Sender of
          {SenderPid, _} ->
              erlang:exit(SenderPid, kill);
          undefined ->
              ok
      end,
      {noreply, NewState#state{req=undefined, req_ctx=undefined}};
  %% ignore #pipe_log for now, since riak_kv_mrc_pipe does not enable it
  
  %% LEGACY Handle response from mapred_stream/mapred_bucket_stream
  