Late updates to ReplicatedChronicleMap throw LateUpdateException

src/main/java/net/openhft/chronicle/hash/replication/LateUpdateException.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.openhft.chronicle.hash.replication;
+
+/**
+ * Thrown if update to replicated {@link net.openhft.chronicle.hash.ChronicleHash} is rejected
+ * due to local time is lesser than the time of the latest update to the entry. This most likely
+ * indicates bugs or largely inconsistent timing across replicating nodes. If this is tolerated,
+ * you should catch {@code LateUpdateException} and ignore it.
+ */
+public class LateUpdateException extends RuntimeException {
+    private static final long serialVersionUID = 0L;
+
+    private final byte originIdentifier;
+    private final long lastUpdateTimestamp;
+    private final byte thisNodeIdentifier;
+    private final long thisUpdateTimestamp;
+
+    public LateUpdateException(byte originIdentifier, long lastUpdateTimestamp,
+                               byte thisNodeIdentifier, long thisUpdateTimestamp) {
+        super("Late update : " +
+                "last update time: " + lastUpdateTimestamp +
+                ", last update origin node identifier: " + originIdentifier +
+                ", time of this update: " + thisUpdateTimestamp +
+                ", identifier of this node: " + thisNodeIdentifier);
+        this.originIdentifier = originIdentifier;
+        this.lastUpdateTimestamp = lastUpdateTimestamp;
+        this.thisNodeIdentifier = thisNodeIdentifier;
+        this.thisUpdateTimestamp = thisUpdateTimestamp;
+    }
+
+    public long thisUpdateTimestamp() {
+        return thisUpdateTimestamp;
+    }
+
+    public byte thisNodeIdentifier() {
+        return thisNodeIdentifier;
+    }
+
+    public long lastUpdateTimestamp() {
+        return lastUpdateTimestamp;
+    }
+
+    public byte lastUpdateOriginNodeIdentifier() {
+        return originIdentifier;
+    }
+}

src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java

@@ -222,9 +222,8 @@ abstract class AbstractChronicleMap<K, V> extends AbstractMap<K, V>
             // 2. Always "try to read" the previous value (not via if (c.containsKey()) {...
             // because BytesChronicleMap relies on it to write null to the output stream
             V prevValue = c.getUsing(null);
-            if (prevValue != null) {
+            if (prevValue != null)
                 c.put(value);
-            }
             return prevValue;
         }
     }

src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java

@@ -19,6 +19,7 @@
 package net.openhft.chronicle.map;
 
 import net.openhft.chronicle.hash.replication.AbstractReplication;
+import net.openhft.chronicle.hash.replication.LateUpdateException;
 import net.openhft.chronicle.hash.replication.TimeProvider;
 import net.openhft.chronicle.hash.serialization.BytesReader;
 import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
@@ -515,11 +516,19 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
             assert identifier != 0 && newIdentifier != 0 :
                     "Own entry replication state or " +
                             "update replication state is not initialized";
-            if (timestamp < newTimestamp)
-                return false;
-            if (timestamp > newTimestamp)
-                return true;
-            return identifier > newIdentifier;
+            boolean shouldIgnore;
+            if (timestamp < newTimestamp) {
+                shouldIgnore = false;
+            } else if (timestamp > newTimestamp) {
+                shouldIgnore = true;
+            } else {
+                shouldIgnore = identifier > newIdentifier;
+            }
+            if (shouldIgnore && !remoteUpdate()) {
+                assert newIdentifier == m().identifier();
+                throw new LateUpdateException(identifier, timestamp, newIdentifier, newTimestamp);
+            }
+            return shouldIgnore;
         }
 
         public void dirtyEntries(final long dirtyFromTimeStamp,