Change ShardFollowTask to reuse common serialization logic (#39094)

Initially in #38910, ShardFollowTask was reusing ImmutableFollowParameters'
serialization logic. After merging, bwc tests failed sometimes and
the binary serialization that ShardFollowTask was originally was using
was added back. ImmutableFollowParameters is using optional fields (optional vint)
while ShardFollowTask was not (vint).

build.gradle

@@ -160,9 +160,8 @@ task verifyVersions {
  * after the backport of the backcompat code is complete.
  */
 
-boolean bwc_tests_enabled = false
-final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/pull/39094" /* place a PR link here when committing bwc changes */
-
+boolean bwc_tests_enabled = true
+final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTask.java

@@ -96,21 +96,15 @@ public static ShardFollowTask readFrom(StreamInput in) throws IOException {
         String remoteCluster = in.readString();
         ShardId followShardId = ShardId.readShardId(in);
         ShardId leaderShardId = ShardId.readShardId(in);
-        // TODO: use ImmutableFollowParameters(StreamInput) constructor
-        int maxReadRequestOperationCount = in.readVInt();
-        ByteSizeValue maxReadRequestSize = new ByteSizeValue(in);
-        int maxOutstandingReadRequests = in.readVInt();
-        int maxWriteRequestOperationCount = in.readVInt();
-        ByteSizeValue maxWriteRequestSize = new ByteSizeValue(in);
-        int maxOutstandingWriteRequests = in.readVInt();
-        int maxWriteBufferCount = in.readVInt();
-        ByteSizeValue maxWriteBufferSize = new ByteSizeValue(in);
-        TimeValue maxRetryDelay = in.readTimeValue();
-        TimeValue readPollTimeout = in.readTimeValue();
-        Map<String, String> headers = Collections.unmodifiableMap(in.readMap(StreamInput::readString, StreamInput::readString));
-        return new ShardFollowTask(remoteCluster, followShardId, leaderShardId, maxReadRequestOperationCount,
-            maxWriteRequestOperationCount, maxOutstandingReadRequests, maxOutstandingWriteRequests, maxReadRequestSize,
-            maxWriteRequestSize, maxWriteBufferCount, maxWriteBufferSize, maxRetryDelay, readPollTimeout, headers);
+        return new ShardFollowTask(remoteCluster, followShardId, leaderShardId, in);
+    }
+
+    private ShardFollowTask(String remoteCluster, ShardId followShardId, ShardId leaderShardId, StreamInput in) throws IOException {
+        super(in);
+        this.remoteCluster = remoteCluster;
+        this.followShardId = followShardId;
+        this.leaderShardId = leaderShardId;
+        this.headers = Collections.unmodifiableMap(in.readMap(StreamInput::readString, StreamInput::readString));
     }
 
     public String getRemoteCluster() {
@@ -139,17 +133,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(remoteCluster);
         followShardId.writeTo(out);
         leaderShardId.writeTo(out);
-        // TODO: use super.writeTo()
-        out.writeVLong(getMaxReadRequestOperationCount());
-        getMaxReadRequestSize().writeTo(out);
-        out.writeVInt(getMaxOutstandingReadRequests());
-        out.writeVLong(getMaxWriteRequestOperationCount());
-        getMaxWriteRequestSize().writeTo(out);
-        out.writeVInt(getMaxOutstandingWriteRequests());
-        out.writeVInt(getMaxWriteBufferCount());
-        getMaxWriteBufferSize().writeTo(out);
-        out.writeTimeValue(getMaxRetryDelay());
-        out.writeTimeValue(getReadPollTimeout());
+        super.writeTo(out);
         out.writeMap(headers, StreamOutput::writeString, StreamOutput::writeString);
     }