ONOS-4396: Fix for EC Map synchronization failing silently due to serialization failures.

With this change we proactively fail map updates when serialization failures can occur and immediately notify the caller

Change-Id: I62a8a84731b9c2a6eeff7fa6f8336dc74234bf30

@@ -339,8 +339,11 @@ public class EventuallyConsistentMapImpl<K, V>
         checkNotNull(value, ERROR_NULL_VALUE);
         checkNotNull(value, ERROR_NULL_VALUE);
 
 
         MapValue<V> newValue = new MapValue<>(value, timestampProvider.apply(key, value));
         MapValue<V> newValue = new MapValue<>(value, timestampProvider.apply(key, value));
+        // Before mutating local map, ensure the update can be serialized without errors.
+        // This prevents replica divergence due to serialization failures.
+        UpdateEntry<K, V> update = serializer.copy(new UpdateEntry<K, V>(key, newValue));
         if (putInternal(key, newValue)) {
         if (putInternal(key, newValue)) {
-            notifyPeers(new UpdateEntry<>(key, newValue), peerUpdateFunction.apply(key, value));
+            notifyPeers(update, peerUpdateFunction.apply(key, value));
             notifyListeners(new EventuallyConsistentMapEvent<>(mapName, PUT, key, value));
             notifyListeners(new EventuallyConsistentMapEvent<>(mapName, PUT, key, value));
         }
         }
     }
     }
@@ -417,13 +420,15 @@ public class EventuallyConsistentMapImpl<K, V>
 
 
         AtomicBoolean updated = new AtomicBoolean(false);
         AtomicBoolean updated = new AtomicBoolean(false);
         AtomicReference<MapValue<V>> previousValue = new AtomicReference<>();
         AtomicReference<MapValue<V>> previousValue = new AtomicReference<>();
-        MapValue<V> computedValue = items.compute(key, (k, mv) -> {
+        MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> {
             previousValue.set(mv);
             previousValue.set(mv);
             V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get());
             V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get());
             MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue));
             MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue));
             if (mv == null || newValue.isNewerThan(mv)) {
             if (mv == null || newValue.isNewerThan(mv)) {
                 updated.set(true);
                 updated.set(true);
-                return newValue;
+                // We return a copy to ensure updates to peers can be serialized.
+                // This prevents replica divergence due to serialization failures.
+                return serializer.copy(newValue);
             } else {
             } else {
                 return mv;
                 return mv;
             }
             }

@@ -78,6 +78,11 @@ public class KryoSerializer implements StoreSerializer {
     }
     }
 
 
     @Override
     @Override
+    public <T> T copy(T object) {
+        return decode(encode(object));
+    }
+
+    @Override
     public String toString() {
     public String toString() {
         return MoreObjects.toStringHelper(getClass())
         return MoreObjects.toStringHelper(getClass())
                 .add("serializerPool", serializerPool)
                 .add("serializerPool", serializerPool)

@@ -75,4 +75,13 @@ public interface StoreSerializer {
      * @param <T> decoded type
      * @param <T> decoded type
      */
      */
     <T> T decode(final InputStream stream);
     <T> T decode(final InputStream stream);
+
+    /**
+     * Returns a copy of the specfied object.
+     *
+     * @param object object to copy
+     * @return a copy of the object
+     * @param <T> object type
+     */
+    <T> T copy(final T object);
 }
 }