Map/Reduce incorrect results with a non-shared non-tx intermediate cache

In a non-tx cache, if a command is started with topology id T, and when it is replicated on another node the distribution interceptor sees topology T+1, it throws an OutdatedTopologyException. The originator of the command will then retry the command, setting topology T+1.

When this happens with a PutKeyValueCommand(k, MapReduceManagerImpl.DeltaAwareList), it can lead to duplicate intermediate values.

Say A is the primary owner of k in T, B is a backup owner both in T and T+1, and C is the backup owner in T and the primary owner in T+1 (i.e. C just joined and a rebalance is in progress during T - see NonTxBackupOwnerBecomingPrimaryOwnerTest).

A starts the PutKeyValueCommand and replicates it to B and C. C applies the command, but B already has topology T+1 and throws an OutdatedTopologyException. A installs topology T+1, sends the command to C (as the new primary owner), which replicates it to B and then applies it locally a second time.

This scenario can happen during a M/R task even without nodes joining or leaving. That's because CreateCacheCommand only calls getCache() on each member, it doesn't wait for the cache to have a certain number of members or for state transfer to be complete for all the members. The last member to join the intermediate cache is guaranteed to have topology T+1, but the others may have topology T by the time the combine phase starts inserting values in the intermediate cache.

I have seen the OutdatedTopologyException happen pretty often during the test suite, especially after I removed the duplicate invokeRemotely call in MapReduceTask.executeTaskInit(). Most of them were harmless, but there was one failure in CI: http://ci.infinispan.org/viewLog.html?buildId=9811&tab=buildResultsDiv&buildTypeId=bt8

A short-term fix would be to wait for all the members to finish joining in CreateCacheCommand. Long-term, M/R tasks should be resilient to topology changes, so we should investigate making PutKeyValue(k, DeltaAwareList) handle OutdatedTopologyException s.