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.

Author: nik9000

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());
             }

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());