diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexAction.java
index 0085f7db9fa08..a71ed64ac33e0 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexAction.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexAction.java
@@ -45,6 +45,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 public class CreateAndFollowIndexAction extends Action<CreateAndFollowIndexAction.Request, CreateAndFollowIndexAction.Response> {
 
@@ -89,6 +90,19 @@ public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             followRequest.writeTo(out);
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return Objects.equals(followRequest, request.followRequest);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(followRequest);
+        }
     }
 
     public static class Response extends ActionResponse implements ToXContentObject {
@@ -145,6 +159,21 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.endObject();
             return builder;
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Response response = (Response) o;
+            return followIndexCreated == response.followIndexCreated &&
+                followIndexShardsAcked == response.followIndexShardsAcked &&
+                indexFollowingStarted == response.indexFollowingStarted;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(followIndexCreated, followIndexShardsAcked, indexFollowingStarted);
+        }
     }
 
     public static class TransportAction extends TransportMasterNodeAction<Request, Response> {
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowIndexAction.java
index d10aa90a5d422..8c58407123c8e 100644
--- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowIndexAction.java
+++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowIndexAction.java
@@ -47,6 +47,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReferenceArray;
@@ -127,6 +128,7 @@ public void readFrom(StreamInput in) throws IOException {
             leaderIndex = in.readString();
             followIndex = in.readString();
             batchSize = in.readVLong();
+            concurrentProcessors = in.readVInt();
             processorMaxTranslogBytes = in.readVLong();
         }
 
@@ -136,8 +138,26 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeString(leaderIndex);
             out.writeString(followIndex);
             out.writeVLong(batchSize);
+            out.writeVInt(concurrentProcessors);
             out.writeVLong(processorMaxTranslogBytes);
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return batchSize == request.batchSize &&
+                concurrentProcessors == request.concurrentProcessors &&
+                processorMaxTranslogBytes == request.processorMaxTranslogBytes &&
+                Objects.equals(leaderIndex, request.leaderIndex) &&
+                Objects.equals(followIndex, request.followIndex);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(leaderIndex, followIndex, batchSize, concurrentProcessors, processorMaxTranslogBytes);
+        }
     }
 
     public static class Response extends AcknowledgedResponse {
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexRequestTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexRequestTests.java
new file mode 100644
index 0000000000000..18900bc852da5
--- /dev/null
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexRequestTests.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.test.AbstractStreamableTestCase;
+
+public class CreateAndFollowIndexRequestTests extends AbstractStreamableTestCase<CreateAndFollowIndexAction.Request> {
+
+    @Override
+    protected CreateAndFollowIndexAction.Request createBlankInstance() {
+        return new CreateAndFollowIndexAction.Request();
+    }
+
+    @Override
+    protected CreateAndFollowIndexAction.Request createTestInstance() {
+        CreateAndFollowIndexAction.Request request = new CreateAndFollowIndexAction.Request();
+        request.setFollowRequest(FollowIndexRequestTests.createTestRequest());
+        return request;
+    }
+}
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexResponseTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexResponseTests.java
new file mode 100644
index 0000000000000..11a518ef06757
--- /dev/null
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/CreateAndFollowIndexResponseTests.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.test.AbstractStreamableTestCase;
+
+public class CreateAndFollowIndexResponseTests extends AbstractStreamableTestCase<CreateAndFollowIndexAction.Response> {
+
+    @Override
+    protected CreateAndFollowIndexAction.Response createBlankInstance() {
+        return new CreateAndFollowIndexAction.Response();
+    }
+
+    @Override
+    protected CreateAndFollowIndexAction.Response createTestInstance() {
+        return new CreateAndFollowIndexAction.Response(randomBoolean(), randomBoolean(), randomBoolean());
+    }
+}
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowIndexRequestTests.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowIndexRequestTests.java
new file mode 100644
index 0000000000000..9a0557b369a77
--- /dev/null
+++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/FollowIndexRequestTests.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.test.AbstractStreamableTestCase;
+
+public class FollowIndexRequestTests extends AbstractStreamableTestCase<FollowIndexAction.Request> {
+
+    @Override
+    protected FollowIndexAction.Request createBlankInstance() {
+        return new FollowIndexAction.Request();
+    }
+
+    @Override
+    protected FollowIndexAction.Request createTestInstance() {
+        return createTestRequest();
+    }
+
+    static FollowIndexAction.Request createTestRequest() {
+        FollowIndexAction.Request request = new FollowIndexAction.Request();
+        request.setLeaderIndex(randomAlphaOfLength(4));
+        request.setFollowIndex(randomAlphaOfLength(4));
+        request.setBatchSize(randomNonNegativeLong());
+        request.setConcurrentProcessors(randomIntBetween(0, Integer.MAX_VALUE));
+        request.setProcessorMaxTranslogBytes(randomNonNegativeLong());
+        return request;
+    }
+}