Ensure local read/write operations are causally consistent when a ConsistentMap …

…is build with weakConsistency option

Change-Id: I1ec96048b646f5074e5e50001e08753212b98395

@@ -26,8 +26,12 @@ import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 
 /**
- * Extension of DefaultAsyncConsistentMap that provides a weaker read consistency
+ * Extension of {@link DefaultAsyncConsistentMap} that provides a weaker read consistency
  * guarantee in return for better read performance.
+ * <p>
+ * For read/write operations that are local to a node this map implementation provides
+ * guarantees similar to a ConsistentMap. However for read/write operations executed
+ * across multiple nodes this implementation only provides eventual consistency.
  *
  * @param <K> key type
  * @param <V> value type
@@ -68,4 +72,10 @@ public class AsyncCachingConsistentMap<K, V> extends DefaultAsyncConsistentMap<K
         }
         return cache.getUnchecked(key);
     }
+
+    @Override
+    protected void beforeUpdate(K key) {
+        super.beforeUpdate(key);
+        cache.invalidate(key);
+    }
 }
\ No newline at end of file

@@ -405,6 +405,14 @@ public class DefaultAsyncConsistentMap<K, V>  implements AsyncConsistentMap<K, V
                 .thenApply(v -> v.updated());
     }
 
+    /**
+     * Pre-update hook for performing required checks/actions before going forward with an update operation.
+     * @param key map key.
+     */
+    protected void beforeUpdate(K key) {
+        checkIfUnmodifiable();
+    }
+
     private Map.Entry<K, Versioned<V>> mapRawEntry(Map.Entry<String, Versioned<byte[]>> e) {
         return Maps.immutableEntry(dK(e.getKey()), e.getValue().<V>map(serializer::decode));
     }
@@ -413,7 +421,7 @@ public class DefaultAsyncConsistentMap<K, V>  implements AsyncConsistentMap<K, V
                                                                Match<V> oldValueMatch,
                                                                Match<Long> oldVersionMatch,
                                                                V value) {
-        checkIfUnmodifiable();
+        beforeUpdate(key);
         return database.mapUpdate(name,
                 keyCache.getUnchecked(key),
                 oldValueMatch.map(serializer::encode),