anchor.diff:  -export([init/1, handle_input/3, handle_input_done/1, handle_event/2,
            handle_sync_event/3, handle_info/2, handle_timeout/1, terminate/2]).
   
  --record(state, {done=false, qterm, fsms=dict:new(), mapper_data=[]}).
  +-record(state, {done=false, qterm, fsms=dict:new(), mapper_data=[], pending=[]}).
   
   init([QTerm]) ->
       {ok, #state{qterm=QTerm}}.
  
positive.code:                                 ClaimLists = riak_kv_mapred_planner:plan_map(NewKeys),
                                  {NewFSMs, _ClaimLists1, FsmKeys} = schedule_input(NewKeys, ClaimLists, QTerm, FSMs, State),
                                  MapperData1 = lists:keydelete(Id, 1, lists:keydelete(Pid, 1, MapperData ++ FsmKeys)),
                                  {no_output, maybe_done(State#state{mapper_data=MapperData1, fsms=NewFSMs})}
                              catch
                                  _:Error ->
                                      {stop, {error, {no_candidate_nodes, Error, erlang:get_stacktrace(), MapperData}}, State}
  
negative.code:                     end;
                  false ->
                      MapperData1 = lists:keydelete(Pid, 1, MapperData),
                      maybe_done(State#state{mapper_data=MapperData1})
              end;
          false ->
              {stop, {error, {dead_mapper, erlang:get_stacktrace(), MapperData}}, State}
  