Fix translog bwc serialization (#36676)

Serializing of a Translog#Index from v7.0.0 to 6.x is broken since
#29224 where we removed the _parent field.

Relates #29224

Author: dnhatn

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -1227,6 +1227,9 @@ private void write(final StreamOutput out) throws IOException {
             out.writeString(type);
             out.writeBytesReference(source);
             out.writeOptionalString(routing);
+            if (format < FORMAT_NO_PARENT) {
+                 out.writeOptionalString(null); // _parent
+            }
             out.writeLong(version);
             if (format < FORMAT_NO_VERSION_TYPE) {
                 out.writeByte(VersionType.EXTERNAL.getValue());

server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -35,6 +35,7 @@
 import org.apache.lucene.util.LineFileDocs;
 import org.apache.lucene.util.LuceneTestCase;
 import org.elasticsearch.Assertions;
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.Strings;
@@ -72,6 +73,7 @@
 import org.elasticsearch.index.translog.Translog.Location;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.IndexSettingsModule;
+import org.elasticsearch.test.VersionUtils;
 import org.hamcrest.Matchers;
 import org.junit.After;
 import org.junit.Before;
@@ -2814,9 +2816,12 @@ public void testTranslogOpSerialization() throws Exception {
         Engine.IndexResult eIndexResult = new Engine.IndexResult(1, randomPrimaryTerm, randomSeqNum, true);
         Translog.Index index = new Translog.Index(eIndex, eIndexResult);
 
+        Version wireVersion = VersionUtils.randomVersionBetween(random(), Version.CURRENT.minimumCompatibilityVersion(), Version.CURRENT);
         BytesStreamOutput out = new BytesStreamOutput();
+        out.setVersion(wireVersion);
         Translog.Operation.writeOperation(out, index);
         StreamInput in = out.bytes().streamInput();
+        in.setVersion(wireVersion);
         Translog.Index serializedIndex = (Translog.Index) Translog.Operation.readOperation(in);
         assertEquals(index, serializedIndex);
 
@@ -2826,8 +2831,10 @@ public void testTranslogOpSerialization() throws Exception {
         Translog.Delete delete = new Translog.Delete(eDelete, eDeleteResult);
 
         out = new BytesStreamOutput();
+        out.setVersion(wireVersion);
         Translog.Operation.writeOperation(out, delete);
         in = out.bytes().streamInput();
+        in.setVersion(wireVersion);
         Translog.Delete serializedDelete = (Translog.Delete) Translog.Operation.readOperation(in);
         assertEquals(delete, serializedDelete);
     }