Binary incompatible serializer changes

- If the field type is fixed and the type is final, Class info can be omitted
- Annotations serializer to use optimization based on the fact Map<String, String> and non-null key/value
- Reduce number of Map copy required for ImmutableMap serializer
- Reduce number of array copy behind Immutable{List, Set} serializer

Change-Id: Ie467a943a33fbfb43b289b8b71ad91ee5890bfb0

@@ -15,11 +15,12 @@
  */
 package org.onosproject.store.device.impl;
 
+import static org.onosproject.store.serializers.DeviceIdSerializer.deviceIdSerializer;
+
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.device.DeviceDescription;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.store.impl.Timestamped;
-
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
@@ -41,7 +42,7 @@ public class InternalDeviceEventSerializer extends Serializer<InternalDeviceEven
     @Override
     public void write(Kryo kryo, Output output, InternalDeviceEvent event) {
         kryo.writeClassAndObject(output, event.providerId());
-        kryo.writeClassAndObject(output, event.deviceId());
+        kryo.writeObject(output, event.deviceId(), deviceIdSerializer());
         kryo.writeClassAndObject(output, event.deviceDescription());
     }
 
@@ -49,7 +50,7 @@ public class InternalDeviceEventSerializer extends Serializer<InternalDeviceEven
     public InternalDeviceEvent read(Kryo kryo, Input input,
                                Class<InternalDeviceEvent> type) {
         ProviderId providerId = (ProviderId) kryo.readClassAndObject(input);
-        DeviceId deviceId = (DeviceId) kryo.readClassAndObject(input);
+        DeviceId deviceId = kryo.readObject(input, DeviceId.class, deviceIdSerializer());
 
         @SuppressWarnings("unchecked")
         Timestamped<DeviceDescription> deviceDescription

@@ -15,9 +15,10 @@
  */
 package org.onosproject.store.device.impl;
 
+import static org.onosproject.store.serializers.DeviceIdSerializer.deviceIdSerializer;
+
 import org.onosproject.net.DeviceId;
 import org.onosproject.store.Timestamp;
-
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
@@ -38,14 +39,14 @@ public class InternalDeviceOfflineEventSerializer extends Serializer<InternalDev
 
     @Override
     public void write(Kryo kryo, Output output, InternalDeviceOfflineEvent event) {
-        kryo.writeClassAndObject(output, event.deviceId());
+        kryo.writeObject(output, event.deviceId(), deviceIdSerializer());
         kryo.writeClassAndObject(output, event.timestamp());
     }
 
     @Override
     public InternalDeviceOfflineEvent read(Kryo kryo, Input input,
                                Class<InternalDeviceOfflineEvent> type) {
-        DeviceId deviceId = (DeviceId) kryo.readClassAndObject(input);
+        DeviceId deviceId = kryo.readObject(input, DeviceId.class, deviceIdSerializer());
         Timestamp timestamp = (Timestamp) kryo.readClassAndObject(input);
 
         return new InternalDeviceOfflineEvent(deviceId, timestamp);

@@ -15,6 +15,8 @@
  */
 package org.onosproject.store.device.impl;
 
+import static org.onosproject.store.serializers.DeviceIdSerializer.deviceIdSerializer;
+
 import java.util.List;
 
 import org.onosproject.net.DeviceId;
@@ -43,7 +45,7 @@ public class InternalPortEventSerializer extends Serializer<InternalPortEvent> {
     @Override
     public void write(Kryo kryo, Output output, InternalPortEvent event) {
         kryo.writeClassAndObject(output, event.providerId());
-        kryo.writeClassAndObject(output, event.deviceId());
+        kryo.writeObject(output, event.deviceId(), deviceIdSerializer());
         kryo.writeClassAndObject(output, event.portDescriptions());
     }
 
@@ -51,7 +53,7 @@ public class InternalPortEventSerializer extends Serializer<InternalPortEvent> {
     public InternalPortEvent read(Kryo kryo, Input input,
                                Class<InternalPortEvent> type) {
         ProviderId providerId = (ProviderId) kryo.readClassAndObject(input);
-        DeviceId deviceId = (DeviceId) kryo.readClassAndObject(input);
+        DeviceId deviceId = kryo.readObject(input, DeviceId.class, deviceIdSerializer());
 
         @SuppressWarnings("unchecked")
         Timestamped<List<PortDescription>> portDescriptions

@@ -15,6 +15,8 @@
  */
 package org.onosproject.store.device.impl;
 
+import static org.onosproject.store.serializers.DeviceIdSerializer.deviceIdSerializer;
+
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.device.PortDescription;
 import org.onosproject.net.provider.ProviderId;
@@ -41,7 +43,7 @@ public class InternalPortStatusEventSerializer extends Serializer<InternalPortSt
     @Override
     public void write(Kryo kryo, Output output, InternalPortStatusEvent event) {
         kryo.writeClassAndObject(output, event.providerId());
-        kryo.writeClassAndObject(output, event.deviceId());
+        kryo.writeObject(output, event.deviceId(), deviceIdSerializer());
         kryo.writeClassAndObject(output, event.portDescription());
     }
 
@@ -49,7 +51,7 @@ public class InternalPortStatusEventSerializer extends Serializer<InternalPortSt
     public InternalPortStatusEvent read(Kryo kryo, Input input,
                                Class<InternalPortStatusEvent> type) {
         ProviderId providerId = (ProviderId) kryo.readClassAndObject(input);
-        DeviceId deviceId = (DeviceId) kryo.readClassAndObject(input);
+        DeviceId deviceId = kryo.readObject(input, DeviceId.class, deviceIdSerializer());
         @SuppressWarnings("unchecked")
         Timestamped<PortDescription> portDescription = (Timestamped<PortDescription>) kryo.readClassAndObject(input);
 

@@ -21,24 +21,48 @@ import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.serializers.DefaultSerializers;
+import com.esotericsoftware.kryo.serializers.DefaultSerializers.StringSerializer;
+import com.esotericsoftware.kryo.serializers.MapSerializer;
 
 import java.util.HashMap;
+import java.util.Map;
 
 public class AnnotationsSerializer extends Serializer<DefaultAnnotations> {
 
+    private static final StringSerializer STR_SERIALIZER
+        = new DefaultSerializers.StringSerializer();
+
+    private static final MapSerializer MAP_SERIALIZER = stringMapSerializer();
+
+    /**
+     * Returns a MapSerializer for {@code Map<String, String>} with
+     * no null key or value.
+     *
+     * @return serializer
+     */
+    private static MapSerializer stringMapSerializer() {
+        MapSerializer serializer = new MapSerializer();
+        serializer.setKeysCanBeNull(false);
+        serializer.setKeyClass(String.class, STR_SERIALIZER);
+        serializer.setValuesCanBeNull(false);
+        serializer.setValueClass(String.class, STR_SERIALIZER);
+        return serializer;
+    }
+
     public AnnotationsSerializer() {
         super(false, true);
     }
 
     @Override
     public void write(Kryo kryo, Output output, DefaultAnnotations object) {
-        kryo.writeObject(output, object.asMap());
+        kryo.writeObject(output, object.asMap(), MAP_SERIALIZER);
     }
 
     @Override
     public DefaultAnnotations read(Kryo kryo, Input input, Class<DefaultAnnotations> type) {
         DefaultAnnotations.Builder b = DefaultAnnotations.builder();
-        HashMap<String, String> map = kryo.readObject(input, HashMap.class);
+        Map<String, String> map = kryo.readObject(input, HashMap.class, MAP_SERIALIZER);
         map.forEach((k, v) -> b.set(k, v));
 
         return b.build();

@@ -36,14 +36,14 @@ public final class DefaultApplicationIdSerializer extends Serializer<DefaultAppl
 
     @Override
     public void write(Kryo kryo, Output output, DefaultApplicationId object) {
-        kryo.writeObject(output, object.id());
-        kryo.writeObject(output, object.name());
+        output.writeShort(object.id());
+        output.writeString(object.name());
     }
 
     @Override
     public DefaultApplicationId read(Kryo kryo, Input input, Class<DefaultApplicationId> type) {
-        short id = kryo.readObject(input, Short.class);
-        String name = kryo.readObject(input, String.class);
+        short id = input.readShort();
+        String name = input.readString();
         return new DefaultApplicationId(id, name);
     }
 }

@@ -27,6 +27,12 @@ import com.esotericsoftware.kryo.io.Output;
 */
 public final class DeviceIdSerializer extends Serializer<DeviceId> {
 
+    private static final DeviceIdSerializer INSTANCE = new DeviceIdSerializer();
+
+    public static final DeviceIdSerializer deviceIdSerializer() {
+        return INSTANCE;
+    }
+
     /**
      * Creates {@link DeviceId} serializer instance.
      */

@@ -20,7 +20,6 @@ import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableList.Builder;
 
 /**
  * Creates {@link ImmutableList} serializer instance.
@@ -44,12 +43,19 @@ public class ImmutableListSerializer extends Serializer<ImmutableList<?>> {
 
     @Override
     public ImmutableList<?> read(Kryo kryo, Input input,
-            Class<ImmutableList<?>> type) {
+                                 Class<ImmutableList<?>> type) {
         final int size = input.readInt();
-        Builder<Object> builder = ImmutableList.builder();
-        for (int i = 0; i < size; ++i) {
-            builder.add(kryo.readClassAndObject(input));
+        switch (size) {
+        case 0:
+            return ImmutableList.of();
+        case 1:
+            return ImmutableList.of(kryo.readClassAndObject(input));
+        default:
+            Object[] elms = new Object[size];
+            for (int i = 0; i < size; ++i) {
+                elms[i] = kryo.readClassAndObject(input);
+            }
+            return ImmutableList.copyOf(elms);
         }
-        return builder.build();
     }
 }

@@ -15,24 +15,20 @@
  */
 package org.onosproject.store.serializers;
 
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.Map.Entry;
 
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
-import com.esotericsoftware.kryo.serializers.MapSerializer;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
 
 /**
 * Kryo Serializer for {@link ImmutableMap}.
 */
 public class ImmutableMapSerializer extends Serializer<ImmutableMap<?, ?>> {
 
-    private final MapSerializer mapSerializer = new MapSerializer();
-
     /**
      * Creates {@link ImmutableMap} serializer instance.
      */
@@ -43,16 +39,31 @@ public class ImmutableMapSerializer extends Serializer<ImmutableMap<?, ?>> {
 
     @Override
     public void write(Kryo kryo, Output output, ImmutableMap<?, ?> object) {
-        // wrapping with unmodifiableMap proxy
-        // to avoid Kryo from writing only the reference marker of this instance,
-        // which will be embedded right before this method call.
-        kryo.writeObject(output, Collections.unmodifiableMap(object), mapSerializer);
+        output.writeInt(object.size());
+        for (Entry<?, ?> e : object.entrySet()) {
+            kryo.writeClassAndObject(output, e.getKey());
+            kryo.writeClassAndObject(output, e.getValue());
+        }
     }
 
     @Override
     public ImmutableMap<?, ?> read(Kryo kryo, Input input,
                                     Class<ImmutableMap<?, ?>> type) {
-        Map<?, ?> map = kryo.readObject(input, HashMap.class, mapSerializer);
-        return ImmutableMap.copyOf(map);
+        final int size = input.readInt();
+        switch (size) {
+        case 0:
+            return ImmutableMap.of();
+        case 1:
+            return ImmutableMap.of(kryo.readClassAndObject(input),
+                                   kryo.readClassAndObject(input));
+
+        default:
+            Builder<Object, Object> builder = ImmutableMap.builder();
+            for (int i = 0; i < size; ++i) {
+                builder.put(kryo.readClassAndObject(input),
+                            kryo.readClassAndObject(input));
+            }
+            return builder.build();
+        }
     }
 }

@@ -15,14 +15,10 @@
  */
 package org.onosproject.store.serializers;
 
-import java.util.ArrayList;
-import java.util.List;
-
 import com.esotericsoftware.kryo.Kryo;
 import com.esotericsoftware.kryo.Serializer;
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
-import com.esotericsoftware.kryo.serializers.CollectionSerializer;
 import com.google.common.collect.ImmutableSet;
 
 /**
@@ -30,26 +26,37 @@ import com.google.common.collect.ImmutableSet;
 */
 public class ImmutableSetSerializer extends Serializer<ImmutableSet<?>> {
 
-    private final CollectionSerializer serializer = new CollectionSerializer();
-
     /**
      * Creates {@link ImmutableSet} serializer instance.
      */
     public ImmutableSetSerializer() {
         // non-null, immutable
         super(false, true);
-        serializer.setElementsCanBeNull(false);
     }
 
     @Override
     public void write(Kryo kryo, Output output, ImmutableSet<?> object) {
-        kryo.writeObject(output, object.asList(), serializer);
+        output.writeInt(object.size());
+        for (Object e : object) {
+            kryo.writeClassAndObject(output, e);
+        }
     }
 
     @Override
     public ImmutableSet<?> read(Kryo kryo, Input input,
                                 Class<ImmutableSet<?>> type) {
-        List<?> elms = kryo.readObject(input, ArrayList.class, serializer);
-        return ImmutableSet.copyOf(elms);
+        final int size = input.readInt();
+        switch (size) {
+        case 0:
+            return ImmutableSet.of();
+        case 1:
+            return ImmutableSet.of(kryo.readClassAndObject(input));
+        default:
+            Object[] elms = new Object[size];
+            for (int i = 0; i < size; ++i) {
+                elms[i] = kryo.readClassAndObject(input);
+            }
+            return ImmutableSet.copyOf(elms);
+        }
     }
 }

@@ -15,6 +15,8 @@
  */
 package org.onosproject.store.serializers;
 
+import static org.onosproject.store.serializers.NodeIdSerializer.nodeIdSerializer;
+
 import org.onosproject.cluster.NodeId;
 import org.onosproject.mastership.MastershipTerm;
 
@@ -26,7 +28,7 @@ import com.esotericsoftware.kryo.io.Output;
 /**
  * Kryo Serializer for {@link org.onosproject.mastership.MastershipTerm}.
  */
-public class MastershipTermSerializer extends Serializer<MastershipTerm> {
+public final class MastershipTermSerializer extends Serializer<MastershipTerm> {
 
     /**
      * Creates {@link MastershipTerm} serializer instance.
@@ -38,14 +40,14 @@ public class MastershipTermSerializer extends Serializer<MastershipTerm> {
 
     @Override
     public MastershipTerm read(Kryo kryo, Input input, Class<MastershipTerm> type) {
-        final NodeId node = (NodeId) kryo.readClassAndObject(input);
+        final NodeId node = kryo.readObjectOrNull(input, NodeId.class, nodeIdSerializer());
         final long term = input.readLong();
         return MastershipTerm.of(node, term);
     }
 
     @Override
     public void write(Kryo kryo, Output output, MastershipTerm object) {
-        kryo.writeClassAndObject(output, object.master());
+        kryo.writeObjectOrNull(output, object.master(), nodeIdSerializer());
         output.writeLong(object.termNumber());
     }
 }

@@ -27,6 +27,12 @@ import org.onosproject.cluster.NodeId;
  */
 public final class NodeIdSerializer extends Serializer<NodeId> {
 
+    private static final NodeIdSerializer INSTANCE = new NodeIdSerializer();
+
+    public static final NodeIdSerializer nodeIdSerializer() {
+        return INSTANCE;
+    }
+
     /**
      * Creates {@link NodeId} serializer instance.
      */

@@ -25,8 +25,7 @@ import com.esotericsoftware.kryo.io.Output;
 /**
  * Serializer for {@link PortNumber}.
  */
-public final class PortNumberSerializer extends
-        Serializer<PortNumber> {
+public final class PortNumberSerializer extends Serializer<PortNumber> {
 
     /**
      * Creates {@link PortNumber} serializer instance.