if the primary replies to the originator as a regular message, it will be ordered with the backup command and it solve the problem when the topology is stable.
if a topology changes while the put if absent, we would need to versioning to keep track with command-invocation-id generate which version (and return value) to decide if the command should be handled or not.

Indeed, fixed.

Dan at point 29, the DC on B should contain v3, shouldn't it? Not that it would change much...

I found another failure mode while trying to work around this issue in NonTxPutIfAbsentDuringLeaveStressTest (I was also trying to merge it with NonTxPutIfAbsentDuringJoinStressTest into a NonTxPutIfAbsentDuringRebalanceStressTest, so the stack traces in the attached log NonTxPutIfAbsentDuringRebalanceStressTest.testPutIfAbsentDuringJoin_1.log.gz don't match master).

Say owners(k) = AB in topology t, and owners(k) = BA in topology t+1. In the following scenario, putIfAbsent fails in thread C-app1 and C-app2 with different values:

1. C-app1: start putIfAbsent(k, v1)
2. C-app1: send putIfAbsent(k, v1) to A (primary)
3. A-remote1: receive putIfAbsent(k, v1)
4. A-remote1: send backup request for putIfAbsent(k, v1) to B
5. A-remote1: write k=v1
6. C-remote1: receive null for putIfAbsent(k, v1)
7. C-app2: start putIfAbsent(k, v2)
8. C-app2: send putIfAbsent(k, v2) to A (primary)
9. A-remote2: receive putIfAbsent(k, v1)
10. A-remote2: read v1 from data container, fail the command
11. C-remote2: receive v1 for putIfAbsent(k, v2)
12. C-app2: return v1 (put was unsuccessful)
13. B-remote1: install topology t+1, B is now primary owner
14. B-remote1: receive backup request for putIfAbsent(k, v1)
15. B-remote1: check topology, send OutdatedTopologyException ack to C
16. C-remote3: install topology t+1
17. C-app3: start putIfAbsent(k, v3)
18. C-app3: send putIfAbsent(k, v3) to B (primary)
19. B-remote3: receive putIfAbsent(k, v3)
20. B-remote3: send backup request for putIfAbsent(k, 3) to A
21. B-remote3: write k=v3
22. C-remote3: receive null for putIfAbsent(k, v3)
23. A-remote3: receive backup request for putIfAbsent(k, v3)
24. A-remote3: write k=v3
25. A-remote3: send backup ack for putIfAbsent(k, v3) to C
26. C-remote3: receive backup ack for putIfAbsent(k, v3)
27. C-app3: return null (put was successful)
28. B-remote1: retry, B is now primary owner
29. B-remote1: read v3 from data container, fail the command
30. C-remote1: receive v3 for putIfAbsent(k, v1)
31. C-app1: return v3 (put was unsuccessful)

I think both this scenario and the previous one are worse than the initial report of seeing null, because we're not respecting the READ_COMMITTED isolation level (a transaction is seeing a value that was never committed).

The situation is even worse when the topology changes: a get after a failed putIfAbsent can return not only null, but also a completely different value.

Say owners(k) = AB, and there is a topology change but the owners of k stay the same. In the following scenario, thread B-app3 first sees putIfAbsent(k, v3) = v2, and then get(k) = v1 (B-appX means application thread X on B, and A-remoteX means remote thread X on A):

1. B-app1: start putIfAbsent(k, v1)
2. B-app1: send putIfAbsent(k, v1) to A (primary)
3. A-remote1: receive putIfAbsent(k, v1)
4. A-remote1: send backup request for putIfAbsent(k, v1) to B
5. B-remote1: receive backup request for putIfAbsent(k, v1)
6. B-remote1: write k=v1
7. B-remote1: send backup ack for putIfAbsent(k, v1)
8. A-remote1: check topology, send OutdatedTopologyException back to B
9. A-app2: start putIfAbsent(k, v2)
10. A-app2: send backup request for putIfAbsent(k, v2) to B
11. A-app2: write k=v2
12. B-app3: start putIfAbsent(k, v3)
13. B-app3: send putIfAbsent(k, v3) to A (primary)
14. A-remote3: receive putIfAbsent(k, v3)
15. A-remote3: read v2 from data container, fail the command
16. B-remote3: receive v2 for putIfAbsent(k, v3)
17. B-app3: return v2 for putIfAbsent(k, v3) (put was unsuccessful)
18. B-app3: start get(k)
19. B-app3: read v1 from local container
20. B-app3: return v1 for get(k)
21. B-remote2: receive putIfAbsent(k, v2) backup command
22. B-remote2: write k=v2
23. B-remote2: send backup ack for putIfAbsent(k, v2)
24. A-remote2: receive backup ack for putIfAbsent(k, v2)
25. A-app2: return null for putIfAbsent(k, v2) (put was successful)
26. B-remote1: receive OutdatedTopologyException for putIfAbsent(k, v1)
27. B-remote1: retry, send putIfAbsent(k, v1) to A (primary)
28. A-remote1: receive putIfAbsent(k, v1)
29. A-remote1: read v2 from data container, fail the command
30. B-remote1: receive v2 for putIfAbsent(k, v1)
31. B-app1: return v2 for putIfAbsent(k, v1) (put was unsuccessful)

This situation can happen with triangle algorithm even without any topology change; as primary does not hold the lock during replication, second putIfAbsent may fail before the first putIfAbsent is executed on backup.

FORCE_WRITE_LOCK doesn't work in non-tx caches.

Possible workaround is to use FORCE_WRITE_LOCK flag for the get() operation.

This is causing a random failure in NonTxPutIfAbsentDuringJoinStressTest.