[CCR] Fix request serialization bug (#34917)

and some parameters that were not set in tests.

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/AutoFollowIT.java

@@ -137,9 +137,6 @@ public class AutoFollowIT extends CcrIntegTestCase {
         if (randomBoolean()) {
             request.setMaxReadRequestOperationCount(randomIntBetween(0, Integer.MAX_VALUE));
         }
-        if (randomBoolean()) {
-            request.setMaxWriteBufferSize(new ByteSizeValue(randomNonNegativeLong()));
-        }
         if (randomBoolean()) {
             request.setMaxReadRequestSize(new ByteSizeValue(randomNonNegativeLong(), ByteSizeUnit.BYTES));
         }
@@ -155,6 +152,9 @@ public class AutoFollowIT extends CcrIntegTestCase {
         if (randomBoolean()) {
             request.setMaxWriteBufferSize(new ByteSizeValue(randomNonNegativeLong(), ByteSizeUnit.BYTES));
         }
+        if (randomBoolean()) {
+            request.setMaxWriteRequestSize(new ByteSizeValue(randomNonNegativeLong()));
+        }
         assertTrue(followerClient().execute(PutAutoFollowPatternAction.INSTANCE, request).actionGet().isAcknowledged());
 
         createLeaderIndex("logs-201901", leaderIndexSettings);

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/PutAutoFollowPatternRequestTests.java

@@ -53,6 +53,15 @@ public class PutAutoFollowPatternRequestTests extends AbstractStreamableXContent
         if (randomBoolean()) {
             request.setMaxRetryDelay(TimeValue.timeValueMillis(500));
         }
+        if (randomBoolean()) {
+            request.setMaxWriteRequestOperationCount(randomIntBetween(0, Integer.MAX_VALUE));
+        }
+        if (randomBoolean()) {
+            request.setMaxWriteBufferSize(new ByteSizeValue(randomNonNegativeLong()));
+        }
+        if (randomBoolean()) {
+            request.setMaxWriteRequestSize(new ByteSizeValue(randomNonNegativeLong()));
+        }
         if (randomBoolean()) {
             request.setMaxReadRequestOperationCount(randomIntBetween(0, Integer.MAX_VALUE));
         }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ResumeFollowActionRequestTests.java

@@ -59,6 +59,12 @@ public class ResumeFollowActionRequestTests extends AbstractStreamableXContentTe
         if (randomBoolean()) {
             request.setMaxWriteBufferCount(randomIntBetween(1, Integer.MAX_VALUE));
         }
+        if (randomBoolean()) {
+            request.setMaxWriteRequestOperationCount(randomIntBetween(1, Integer.MAX_VALUE));
+        }
+        if (randomBoolean()) {
+            request.setMaxWriteRequestSize(new ByteSizeValue(randomNonNegativeLong()));
+        }
         if (randomBoolean()) {
             request.setMaxWriteBufferSize(new ByteSizeValue(randomNonNegativeLong(), ByteSizeUnit.BYTES));
         }

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

@@ -266,6 +266,8 @@ public final class ResumeFollowAction extends Action<AcknowledgedResponse> {
             maxReadRequestOperationCount = in.readOptionalVInt();
             maxOutstandingReadRequests = in.readOptionalVInt();
             maxReadRequestSize = in.readOptionalWriteable(ByteSizeValue::new);
+            maxWriteRequestOperationCount = in.readOptionalVInt();
+            maxWriteRequestSize = in.readOptionalWriteable(ByteSizeValue::new);
             maxOutstandingWriteRequests = in.readOptionalVInt();
             maxWriteBufferCount = in.readOptionalVInt();
             maxWriteBufferSize = in.readOptionalWriteable(ByteSizeValue::new);
@@ -280,6 +282,8 @@ public final class ResumeFollowAction extends Action<AcknowledgedResponse> {
             out.writeOptionalVInt(maxReadRequestOperationCount);
             out.writeOptionalVInt(maxOutstandingReadRequests);
             out.writeOptionalWriteable(maxReadRequestSize);
+            out.writeOptionalVInt(maxWriteRequestOperationCount);
+            out.writeOptionalWriteable(maxWriteRequestSize);
             out.writeOptionalVInt(maxOutstandingWriteRequests);
             out.writeOptionalVInt(maxWriteBufferCount);
             out.writeOptionalWriteable(maxWriteBufferSize);