Fix for Portable Serialization incompatible class definition when nes…

…ted portables used

Portable serialization is mistakenly reporting incompatible class-definitions when nested portables are used in certain scenarios.
Details of failed scenario are as follows:
Let's say, we have two Portables: `Parent` and `Child`. `Child` is a field of `Parent` class.
Let's say we have two nodes. Node A and Node B. And both nodes are using a non-default both same portable version.
- `Child` is serialized in Node B. This results with a class definition cached in Node B.
- `Parent` is serialized in Node A and send to Node B.
- When Node B is deserializing `Parent`, it reads the class definition of both `Parent` and `Child` from the wire.
- In Node B, when reading the class definition of `Child`, the version of fields of `Child` class is read wrong.
- We end up with the following exception when checking if the new class definition of `Child` is same as the cached one in Node B
```
Exception in thread "main" com.hazelcast.nio.serialization.HazelcastSerializationException: Incompatible class-definitions with same class-id:
ClassDefinition{factoryId=1, classId=2, version=6, fieldDefinitions=[FieldDefinitionImpl{index=0, fieldName='name', type=UTF, classId=0, factoryId=0, version=0}]} VS
ClassDefinition{factoryId=1, classId=2, version=6, fieldDefinitions=[FieldDefinitionImpl{index=0, fieldName='name', type=UTF, classId=0, factoryId=0, version=6}]}
	at com.hazelcast.internal.serialization.impl.PortableContextImpl$ClassDefinitionContext.register(PortableContextImpl.java:281)
	at com.hazelcast.internal.serialization.impl.PortableContextImpl.registerClassDefinition(PortableContextImpl.java:165)
	at com.hazelcast.internal.serialization.impl.PortableContextImpl.readClassDefinition(PortableContextImpl.java:158)
	at com.hazelcast.internal.serialization.impl.PortableContextImpl.readClassDefinition(PortableContextImpl.java:135)
	at com.hazelcast.internal.serialization.impl.PortableSerializer.setupPositionAndDefinition(PortableSerializer.java:156)
	at com.hazelcast.internal.serialization.impl.PortableSerializer.createReader(PortableSerializer.java:166)
	at com.hazelcast.internal.serialization.impl.PortableSerializer.read(PortableSerializer.java:90)
	at com.hazelcast.internal.serialization.impl.PortableSerializer.read(PortableSerializer.java:81)
	at com.hazelcast.internal.serialization.impl.PortableSerializer.read(PortableSerializer.java:34)
	at com.hazelcast.internal.serialization.impl.StreamSerializerAdapter.read(StreamSerializerAdapter.java:48)
	at com.hazelcast.internal.serialization.impl.AbstractSerializationService.toObject(AbstractSerializationService.java:191)

```

This pr contains both fix and some cleanups.
Fix is to use class version in fields when reading class definition from wire.
In a couple places, fields are made private/final where applicable.

fixes hazelcast#12733

hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/ClassDefinitionImpl.java

@@ -27,8 +27,8 @@
 
 public class ClassDefinitionImpl implements ClassDefinition {
 
-    private int factoryId;
-    private int classId;
+    private final int factoryId;
+    private final int classId;
     private int version = -1;
     private final Map<String, FieldDefinition> fieldDefinitionsMap = new LinkedHashMap<String, FieldDefinition>();
 

hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/FieldDefinitionImpl.java

@@ -21,12 +21,12 @@
 
 public class FieldDefinitionImpl implements FieldDefinition {
 
-    int index;
-    String fieldName;
-    FieldType type;
-    int classId;
-    int factoryId;
-    int version;
+    private final int index;
+    private final String fieldName;
+    private final FieldType type;
+    private final int classId;
+    private final int factoryId;
+    private final int version;
 
     public FieldDefinitionImpl(int index, String fieldName, FieldType type, int version) {
         this(index, fieldName, type, 0, 0, version);

hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/PortableContextImpl.java

@@ -120,7 +120,7 @@ ClassDefinition readClassDefinition(BufferObjectDataInput in, int factoryId, int
             String name = new String(chars);
             int fieldFactoryId = 0;
             int fieldClassId = 0;
-            int fieldVersion = 0;
+            int fieldVersion = version;
             if (type == FieldType.PORTABLE) {
                 // is null
                 if (in.readBoolean()) {

hazelcast/src/test/java/com/hazelcast/nio/serialization/PortableVersionTest.java

@@ -0,0 +1,168 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.nio.serialization;
+
+import com.hazelcast.internal.serialization.SerializationServiceBuilder;
+import com.hazelcast.internal.serialization.impl.DefaultSerializationServiceBuilder;
+import com.hazelcast.spi.serialization.SerializationService;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertEquals;
+
+public class PortableVersionTest {
+
+    public static class Child implements Portable {
+
+        private String name;
+
+        public Child() {
+        }
+
+        public Child(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public int getFactoryId() {
+            return 1;
+        }
+
+        @Override
+        public int getClassId() {
+            return 2;
+        }
+
+        @Override
+        public void writePortable(PortableWriter writer) throws IOException {
+            writer.writeUTF("name", name);
+        }
+
+        @Override
+        public void readPortable(PortableReader reader) throws IOException {
+            name = reader.readUTF("name");
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            Child child = (Child) o;
+
+            return name != null ? name.equals(child.name) : child.name == null;
+        }
+
+        @Override
+        public int hashCode() {
+            return name != null ? name.hashCode() : 0;
+        }
+    }
+
+    public static class Parent implements Portable {
+
+        private Child child;
+
+        public Parent() {
+        }
+
+        public Parent(Child child) {
+            this.child = child;
+        }
+
+        @Override
+        public int getFactoryId() {
+            return 1;
+        }
+
+        @Override
+        public int getClassId() {
+            return 1;
+        }
+
+        @Override
+        public void writePortable(PortableWriter writer) throws IOException {
+            writer.writePortable("child", child);
+        }
+
+        @Override
+        public void readPortable(PortableReader reader) throws IOException {
+            child = reader.readPortable("child");
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            Parent parent = (Parent) o;
+
+            return child != null ? child.equals(parent.child) : parent.child == null;
+        }
+
+        @Override
+        public int hashCode() {
+            return child != null ? child.hashCode() : 0;
+        }
+    }
+
+    // Test for issue https://github.com/hazelcast/hazelcast/issues/12733
+    @Test
+    public void test_nestedPortable_versionedSerializer() {
+        SerializationServiceBuilder builder1 = new DefaultSerializationServiceBuilder();
+        builder1.setPortableVersion(6);
+        builder1.addPortableFactory(1, new PortableFactory() {
+            @Override
+            public Portable create(int classId) {
+                if (classId == 1) {
+                    return new Parent();
+                } else if (classId == 2) {
+                    return new Child();
+                }
+                return null;
+            }
+        });
+        SerializationService ss1 = builder1.build();
+
+
+        SerializationServiceBuilder builder2 = new DefaultSerializationServiceBuilder();
+        builder2.setPortableVersion(6);
+        builder2.addPortableFactory(1, new PortableFactory() {
+            @Override
+            public Portable create(int classId) {
+                if (classId == 1) {
+                    return new Parent();
+                } else if (classId == 2) {
+                    return new Child();
+                }
+                return null;
+            }
+        });
+        SerializationService ss2 = builder2.build();
+
+        //make sure ss2 cached class definition of Child
+        ss2.toData(new Child("sancar"));
+
+        //serialized parent from ss1
+        Parent parent = new Parent(new Child("sancar"));
+        Data data = ss1.toData(parent);
+
+        // cached class definition of Child and the class definition from data coming from ss1 should be compatible
+        assertEquals(parent, ss2.toObject(data));
+    }
+}