ONOS-2495: Workaround for the fact that kryo deserialization is not thread safe

Change-Id: Id1bd3334f9a5c122984d08f97dbbbf622b27cf33

@@ -32,7 +32,6 @@ import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.Versioned;
 import org.onosproject.store.service.Versioned;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 
 
-
 import java.util.Collection;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Map.Entry;
@@ -122,10 +121,17 @@ public class DefaultAsyncConsistentMap<K, V>  implements AsyncConsistentMap<K, V
         this.purgeOnUninstall = purgeOnUninstall;
         this.purgeOnUninstall = purgeOnUninstall;
         this.database.registerConsumer(update -> {
         this.database.registerConsumer(update -> {
             SharedExecutors.getSingleThreadExecutor().execute(() -> {
             SharedExecutors.getSingleThreadExecutor().execute(() -> {
+                if (listeners.isEmpty()) {
+                    return;
+                }
+                try {
                     if (update.target() == MAP_UPDATE) {
                     if (update.target() == MAP_UPDATE) {
                         Result<UpdateResult<String, byte[]>> result = update.output();
                         Result<UpdateResult<String, byte[]>> result = update.output();
                         if (result.success() && result.value().mapName().equals(name)) {
                         if (result.success() && result.value().mapName().equals(name)) {
-                        MapEvent<K, V> mapEvent = result.value().<K, V>map(this::dK, serializer::decode).toMapEvent();
+                            MapEvent<K, V> mapEvent = result.value()
+                                                            .<K, V>map(this::dK,
+                                                                       v -> serializer.decode(Tools.copyOf(v)))
+                                                            .toMapEvent();
                             notifyListeners(mapEvent);
                             notifyListeners(mapEvent);
                         }
                         }
                     } else if (update.target() == TX_COMMIT) {
                     } else if (update.target() == TX_COMMIT) {
@@ -133,12 +139,18 @@ public class DefaultAsyncConsistentMap<K, V>  implements AsyncConsistentMap<K, V
                         if (response.success()) {
                         if (response.success()) {
                             response.updates().forEach(u -> {
                             response.updates().forEach(u -> {
                                 if (u.mapName().equals(name)) {
                                 if (u.mapName().equals(name)) {
-                                MapEvent<K, V> mapEvent = u.<K, V>map(this::dK, serializer::decode).toMapEvent();
+                                    MapEvent<K, V> mapEvent =
+                                            u.<K, V>map(this::dK,
+                                                        v -> serializer.decode(Tools.copyOf(v)))
+                                             .toMapEvent();
                                     notifyListeners(mapEvent);
                                     notifyListeners(mapEvent);
                                 }
                                 }
                             });
                             });
                         }
                         }
                     }
                     }
+                } catch (Exception e) {
+                    log.warn("Error notifying listeners", e);
+                }
             });
             });
         });
         });
         this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
         this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);

@@ -190,6 +190,16 @@ public abstract class Tools {
     }
     }
 
 
     /**
     /**
+     * Returns a copy of the input byte array.
+     *
+     * @param original input
+     * @return copy of original
+     */
+    public static byte[] copyOf(byte[] original) {
+        return Arrays.copyOf(original, original.length);
+    }
+
+    /**
      * Get property as a string value.
      * Get property as a string value.
      *
      *
      * @param properties   properties to be looked up
      * @param properties   properties to be looked up