Fix serialization bug in ShardFollowTask after cutting this class over to

extend from ImmutableFollowParameters.

Prior to #38910 readVInt and writeVInt was used, #38910 accidentally
changed ShardFollowTask to use readOptionalVInt / writeVOptionalVInt.
This changes changes ShardFollowTask back to use vint.

Author: martijnvg

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

@@ -93,15 +93,24 @@ public class ShardFollowTask extends ImmutableFollowParameters implements XPackP
     }
 
     public static ShardFollowTask readFrom(StreamInput in) throws IOException {
-        return new ShardFollowTask(in.readString(), ShardId.readShardId(in), ShardId.readShardId(in), 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));
+        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);
     }
 
     public String getRemoteCluster() {
@@ -130,7 +139,17 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(remoteCluster);
         followShardId.writeTo(out);
         leaderShardId.writeTo(out);
-        super.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());
         out.writeMap(headers, StreamOutput::writeString, StreamOutput::writeString);
     }