Fixing the custom object serialization bug in diffable utils.

server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java

@@ -474,8 +474,10 @@ public void writeTo(StreamOutput out) throws IOException {
             }
             out.writeVInt(upsertsCount);
             for (Map.Entry<K, T> entry : upserts.entrySet()) {
-                keySerializer.writeKey(entry.getKey(), out);
-                valueSerializer.write(entry.getValue(), out);
+                if(valueSerializer.supportsVersion(entry.getValue(), version)) {
+                    keySerializer.writeKey(entry.getKey(), out);
+                    valueSerializer.write(entry.getValue(), out);
+                }
             }
         }
     }

server/src/test/java/org/elasticsearch/cluster/serialization/ClusterSerializationTests.java

@@ -20,11 +20,14 @@
 package org.elasticsearch.cluster.serialization;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.ClusterModule;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterState.Custom;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.ESAllocationTestCase;
+import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.RestoreInProgress;
 import org.elasticsearch.cluster.SnapshotDeletionsInProgress;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -39,14 +42,17 @@
 import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.test.VersionUtils;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
@@ -214,4 +220,142 @@ public void testObjectReuseWhenApplyingClusterStateDiff() throws Exception {
         assertSame("template", serializedClusterState2.metaData().templates().get("test-template"),
             serializedClusterState3.metaData().templates().get("test-template"));
     }
+
+    public static class TestCustomOne extends AbstractNamedDiffable<Custom> implements Custom {
+
+        public static final String TYPE = "test_custom_one";
+        private final String strObject;
+
+        public TestCustomOne(String strObject) {
+            this.strObject = strObject;
+        }
+
+        public TestCustomOne(StreamInput in) throws IOException {
+            this.strObject = in.readString();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeString(strObject);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            {
+                builder.field("custom_string_object", strObject);
+            }
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public String getWriteableName() {
+            return TYPE;
+        }
+
+        public static NamedDiff<Custom> readDiffFrom(StreamInput in) throws IOException {
+            return readDiffFrom(Custom.class, TYPE, in);
+        }
+
+        @Override
+        public Version getMinimalSupportedVersion() {
+            return Version.CURRENT;
+        }
+
+    }
+
+    public static class TestCustomTwo extends AbstractNamedDiffable<Custom>  implements Custom {
+
+        public static final String TYPE = "test_custom_two";
+        private final Integer intObject;
+
+        public TestCustomTwo(Integer intObject) {
+            this.intObject = intObject;
+        }
+
+        public TestCustomTwo(StreamInput in) throws IOException {
+            this.intObject = in.readInt();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeInt(intObject);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            {
+                builder.field("custom_integer_object", intObject);
+            }
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public String getWriteableName() {
+            return TYPE;
+        }
+
+        public static NamedDiff<Custom> readDiffFrom(StreamInput in) throws IOException {
+            return readDiffFrom(Custom.class, TYPE, in);
+        }
+
+        @Override
+        public Version getMinimalSupportedVersion() {
+            return Version.CURRENT.minimumCompatibilityVersion();
+        }
+
+    }
+
+    public void testCustomSerialization() throws Exception {
+        ClusterState.Builder builder = ClusterState.builder(ClusterState.EMPTY_STATE)
+            .putCustom(TestCustomOne.TYPE, new TestCustomOne("test_custom_one"))
+            .putCustom(TestCustomTwo.TYPE, new TestCustomTwo(10));
+
+        ClusterState clusterState = builder.incrementVersion().build();
+
+        Diff<ClusterState> diffs = clusterState.diff(ClusterState.EMPTY_STATE);
+
+        // Add the new customs to named writeables
+        final List<NamedWriteableRegistry.Entry> entries = ClusterModule.getNamedWriteables();
+        entries.add(new NamedWriteableRegistry.Entry(ClusterState.Custom.class, TestCustomOne.TYPE, TestCustomOne::new));
+        entries.add(new NamedWriteableRegistry.Entry(NamedDiff.class, TestCustomOne.TYPE, TestCustomOne::readDiffFrom));
+        entries.add(new NamedWriteableRegistry.Entry(ClusterState.Custom.class, TestCustomTwo.TYPE, TestCustomTwo::new));
+        entries.add(new NamedWriteableRegistry.Entry(NamedDiff.class, TestCustomTwo.TYPE, TestCustomTwo::readDiffFrom));
+
+        // serialize with current version
+        BytesStreamOutput outStream = new BytesStreamOutput();
+        Version version = Version.CURRENT;
+        outStream.setVersion(version);
+        diffs.writeTo(outStream);
+        StreamInput inStream = outStream.bytes().streamInput();
+
+        inStream = new NamedWriteableAwareStreamInput(inStream, new NamedWriteableRegistry(entries));
+        inStream.setVersion(version);
+        Diff<ClusterState> serializedDiffs = ClusterState.readDiffFrom(inStream, clusterState.nodes().getLocalNode());
+        ClusterState stateAfterDiffs = serializedDiffs.apply(ClusterState.EMPTY_STATE);
+
+        // Current version - Both the customs are non null
+        assertThat(stateAfterDiffs.custom(TestCustomOne.TYPE), notNullValue());
+        assertThat(stateAfterDiffs.custom(TestCustomTwo.TYPE), notNullValue());
+
+        // serialize with minimum compatibile version
+        outStream = new BytesStreamOutput();
+        version = Version.CURRENT.minimumCompatibilityVersion();
+        outStream.setVersion(version);
+        diffs.writeTo(outStream);
+        inStream = outStream.bytes().streamInput();
+
+        inStream = new NamedWriteableAwareStreamInput(inStream, new NamedWriteableRegistry(entries));
+        inStream.setVersion(version);
+        serializedDiffs = ClusterState.readDiffFrom(inStream, clusterState.nodes().getLocalNode());
+        stateAfterDiffs = serializedDiffs.apply(ClusterState.EMPTY_STATE);
+
+        // Old version - TestCustomOne is null and TestCustomTwo is not null
+        assertThat(stateAfterDiffs.custom(TestCustomOne.TYPE), nullValue());
+        assertThat(stateAfterDiffs.custom(TestCustomTwo.TYPE), notNullValue());
+    }
+
 }