Fix serialization bug for aggs

I created this bug today in elastic#53793. When a `DelayableWriteable` that references an existing object serializes itself it wasn't taking the version of the node on the other side of the wire into account. This fixes that.
nik9000 · Mar 23, 2020 · cc307c8 · cc307c8
1 parent b9bfba2
commit cc307c8
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 5 deletions.
diff --git a/server/src/main/java/org/elasticsearch/common/io/stream/DelayableWriteable.java b/server/src/main/java/org/elasticsearch/common/io/stream/DelayableWriteable.java
@@ -19,12 +19,12 @@
 
 package org.elasticsearch.common.io.stream;
 
-import org.elasticsearch.Version;
-import org.elasticsearch.common.bytes.BytesReference;
-
 import java.io.IOException;
 import java.util.function.Supplier;
 
+import org.elasticsearch.Version;
+import org.elasticsearch.common.bytes.BytesReference;
+
 /**
  * A holder for {@link Writeable}s that can delays reading the underlying
  * {@linkplain Writeable} when it is read from a remote node.
@@ -60,6 +60,7 @@ private static class Referencing<T extends Writeable> extends DelayableWriteable
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             try (BytesStreamOutput buffer = new BytesStreamOutput()) {
+                buffer.setVersion(out.getVersion());
                 reference.writeTo(buffer);
                 out.writeBytesReference(buffer.bytes());
             }

diff --git a/server/src/test/java/org/elasticsearch/common/io/stream/DelayableWriteableTests.java b/server/src/test/java/org/elasticsearch/common/io/stream/DelayableWriteableTests.java
@@ -30,7 +30,7 @@
 
 public class DelayableWriteableTests extends ESTestCase {
     // NOTE: we don't use AbstractWireSerializingTestCase because we don't implement equals and hashCode.
-    public static class Example implements NamedWriteable {
+    private static class Example implements NamedWriteable {
         private final String s;
 
         public Example(String s) {
@@ -66,7 +66,7 @@ public int hashCode() {
         }
     }
 
-    public static class NamedHolder implements Writeable {
+    private static class NamedHolder implements Writeable {
         private final Example e;
 
         public NamedHolder(Example e) {
@@ -97,6 +97,23 @@ public int hashCode() {
         }
     }
 
+    private static class SneakOtherSideVersionOnWire implements Writeable {
+        private final Version version;
+
+        public SneakOtherSideVersionOnWire() {
+            version = Version.CURRENT;
+        }
+
+        public SneakOtherSideVersionOnWire(StreamInput in) throws IOException {
+            version = Version.readVersion(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            Version.writeVersion(out.getVersion(), out);
+        }
+    }
+
     public void testRoundTripFromReferencing() throws IOException {
         Example e = new Example(randomAlphaOfLength(5));
         DelayableWriteable<Example> original = DelayableWriteable.referencing(e);
@@ -139,6 +156,12 @@ public void testRoundTripFromDelayedFromOldVersionWithNamedWriteable() throws IO
         roundTripTestCase(original, NamedHolder::new);
     }
 
+    public void testSerializesWithRemoteVersion() throws IOException {
+        Version remoteVersion = VersionUtils.randomCompatibleVersion(random(), Version.CURRENT);
+        DelayableWriteable<SneakOtherSideVersionOnWire> original = DelayableWriteable.referencing(new SneakOtherSideVersionOnWire());
+        assertThat(roundTrip(original, SneakOtherSideVersionOnWire::new, remoteVersion).get().version, equalTo(remoteVersion));
+    }
+
     private <T extends Writeable> void roundTripTestCase(DelayableWriteable<T> original, Writeable.Reader<T> reader) throws IOException {
         DelayableWriteable<T> roundTripped = roundTrip(original, reader, Version.CURRENT);
         assertTrue(roundTripped.isDelayed());