Fix translog bwc serialization

[dnhatn]

@martijnvg found that translog is not serialized correctly while working an upgrade test for CCR. Serializing of a Translog#Index from v7.0.0 to 6.x is broken since #29224 where we removed the _parent field. Our tests do not uncover this issue because NodeVersionAllocationDecider prevents allocating a replica on an old version if the primary is on a new version. However, this issue can happen in the following scenarios:

1. A replica on 7.0 gets promoted, and there's another replica on 6.x. Primary-replica resync will ship translog operations from 7.0 to 6.x.
2. A leader cluster is on 7.0, and a follower cluster is on 6.x.
3. ShardFollowTask is running on 7.0, and follower shard copies are on 6.x.

This change fixes the serialization issue and adds a test for the first scenario. The last two situations should be covered in #36648.

Relates #29224

I marked this non-issue for it is an unreleased bug.

[elasticmachine]

Pinging @elastic/es-distributed

[bleskes]

code change LGTM. I'm wondering about the extra complexity of the test.

[bleskes]

(why) Isn't this potentially covered but the 3 node rolling upgrade tests where you upgrade one node, then the second and if that has a primary it will promote the first shard which will cause a resync with the replica. If that's true, do we really need this test?

[dnhatn]

I wrote a rolling upgrade test but was unable to make it fail. I'll give it another shot today.

[dnhatn]

@bleskes The reason that the rolling-upgrade did not fail because (1) primary-replica resync only operations after the global checkpoint, (2) we sync the global checkpoint for every 30 seconds. These make the resync a noop.
Since the newly added test is complicated, should we remove it and let the CCR upgrade-tests verify this BWC issue? WDYT?

[bleskes]

Thanks for checking. I think it's good to have a rolling upgrade test that properly tests all aspects of primary promotions. Maybe we should slow down the GCP sync through a settings (and add concurrent indexing). This may get tricky too but I think we should explore it. @ywelsch any thoughts on this?

[ywelsch]

Writing these tests is tricky, and the REST tests might not provide the best place for doing so. I'm not sure if the 30 second global-checkpoint sync is what's making this difficult to test with a rolling-upgrade (we send the updated global checkpoint to the replicas as soon as an indexing operation completes, and don't wait for 30 seconds). I think the difficult bit is rather the need for continuous indexing while the node with the primary is shutting down, so that the resync will actually have something meaningful to sync. There is currently no way to do this using the REST tests (i.e. index stuff while a node is shutting down).

In this particular instance, I wonder if a unit test would suffice, namely one that follows the likes of ClusterSerializationTests.testSnapshotDeletionsInProgressSerialization which serializes a cluster state with a certain stream version, and then decodes again using that same version. This could be done for the translog operations as a minor extension of testTranslogOpSerialization:

diff --git a/server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java b/server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java
index 7e73f9ef517..ac07a226a25 100644
--- a/server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java
+++ b/server/src/test/java/org/elasticsearch/index/translog/TranslogTests.java
@@ -35,6 +35,7 @@ import org.apache.lucene.store.MockDirectoryWrapper;
 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.shard.ShardId;
 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;
@@ -2815,8 +2817,11 @@ public class TranslogTests extends ESTestCase {
         Translog.Index index = new Translog.Index(eIndex, eIndexResult);
 
         BytesStreamOutput out = new BytesStreamOutput();
+        Version version = VersionUtils.randomVersionBetween(random(), Version.CURRENT.minimumCompatibilityVersion(), Version.CURRENT);
+        out.setVersion(version);
         Translog.Operation.writeOperation(out, index);
         StreamInput in = out.bytes().streamInput();
+        in.setVersion(version);
         Translog.Index serializedIndex = (Translog.Index) Translog.Operation.readOperation(in);
         assertEquals(index, serializedIndex);
 
@@ -2826,8 +2831,10 @@ public class TranslogTests extends ESTestCase {
         Translog.Delete delete = new Translog.Delete(eDelete, eDeleteResult);
 
         out = new BytesStreamOutput();
+        out.setVersion(version);
         Translog.Operation.writeOperation(out, delete);
         in = out.bytes().streamInput();
+        in.setVersion(version);
         Translog.Delete serializedDelete = (Translog.Delete) Translog.Operation.readOperation(in);
         assertEquals(delete, serializedDelete);
     }

Combined with the test that Martijn will add for CCR, this will hopefully give us enough coverage.

[dnhatn]

@bleskes and @ywelsch I think we should be good with a unit test and the tests that @martijnvg will add for CCR. I replaced the mixed-cluster test with the unit test. Would you please have another look? Thank you!

[ywelsch]

LGTM

[dnhatn]

Thanks @bleskes and @ywelsch.