anchor.diff:     {mr, next_phase_data, JobId, AckRange, Data, Round}, _Parents}, State) ->
       %% processing of data for next phase.
       %% save data and send ack
  -    ?TRACE("mr_~s on ~p: received next phase data for ~p: ~p~n",
  -           [JobId, self(), Interval, Data]),
  +    ?TRACE("mr_~s on ~p: received next phase data (round ~p) interval ~p: ~p~n",
  +           [JobId, self(), Round, Interval, Data]),
       NewMRState = mr_state:add_data_to_phase(dht_node_state:get_mr_state(State,
                                                                               JobId),
  -                                                 Data, Interval, Round),
  +                                                 Data, Interval, Round + 1),
       %% send ack with delivery interval
       bulkowner:issue_bulk_owner(uid:get_global_uid(), AckRange, {mr,
                                                                   next_phase_data_ack,
  -                                                               Interval, JobId}),
  +                                                               Interval, JobId,
  +                                                               Round}),
       dht_node_state:set_mr_state(State, JobId, NewMRState);
   
  -on({mr, next_phase_data_ack, AckInterval, JobId, DeliveryInterval}, State) ->
  +on({mr, next_phase_data_ack, AckInterval, JobId, Round, DeliveryInterval}, State) ->
       %% ack from other mr nodes.
       %% check if the whole interval waas acked. If so inform master, wait
       %% otherwise.
  
positive.code:             Ref = uid:get_global_uid(),
              bulkowner:issue_bulk_distribute(Ref, dht_node,
                                              5, {mr, next_phase_data, JobId,
                                                  Range, '_', Round + 1},
                                              {ets, Data});
          _ ->
              ?TRACE("jobs last phase done...sending to client~n", []),
  
negative.code:                 false ->
                      %% io:format("no data for phase...done...~p informs master~n", [self()]),
                      MasterId = mr_state:get(MRState, master_id),
                      api_dht_raw:unreliable_lookup(MasterId, {mr_master, JobId,
                                                               phase_completed,
                                                               Round, AckInterval});
                  _ ->
                      %% io:format("last phase and no data ~p~n", [Round]),
                      MasterId = mr_state:get(MRState, master_id),
  