[CCR] Made leader cluster required in shard follow task.

Left over from #34580

Author: martijnvg

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

@@ -122,7 +122,7 @@ public class ShardFollowTask implements XPackPlugin.XPackPersistentTaskParams {
     }
 
     public ShardFollowTask(StreamInput in) throws IOException {
-        this.leaderCluster = in.readOptionalString();
+        this.leaderCluster = in.readString();
         this.followShardId = ShardId.readShardId(in);
         this.leaderShardId = ShardId.readShardId(in);
         this.maxBatchOperationCount = in.readVInt();
@@ -190,7 +190,7 @@ public String getWriteableName() {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        out.writeOptionalString(leaderCluster);
+        out.writeString(leaderCluster);
         followShardId.writeTo(out);
         leaderShardId.writeTo(out);
         out.writeVLong(maxBatchOperationCount);
@@ -210,9 +210,7 @@ public static ShardFollowTask fromXContent(XContentParser parser) {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
-        if (leaderCluster != null) {
-            builder.field(LEADER_CLUSTER_FIELD.getPreferredName(), leaderCluster);
-        }
+        builder.field(LEADER_CLUSTER_FIELD.getPreferredName(), leaderCluster);
         builder.field(FOLLOW_SHARD_INDEX_FIELD.getPreferredName(), followShardId.getIndex().getName());
         builder.field(FOLLOW_SHARD_INDEX_UUID_FIELD.getPreferredName(), followShardId.getIndex().getUUID());
         builder.field(FOLLOW_SHARD_SHARDID_FIELD.getPreferredName(), followShardId.id());

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

@@ -67,15 +67,6 @@ public ShardFollowTasksExecutor(Settings settings, Client client, ThreadPool thr
 
     @Override
     public void validate(ShardFollowTask params, ClusterState clusterState) {
-        if (params.getLeaderCluster() == null) {
-            // We can only validate IndexRoutingTable in local cluster,
-            // for remote cluster we would need to make a remote call and we cannot do this here.
-            IndexRoutingTable routingTable = clusterState.getRoutingTable().index(params.getLeaderShardId().getIndex());
-            if (routingTable.shard(params.getLeaderShardId().id()).primaryShard().started() == false) {
-                throw new IllegalArgumentException("Not all copies of leader shard are started");
-            }
-        }
-
         IndexRoutingTable routingTable = clusterState.getRoutingTable().index(params.getFollowShardId().getIndex());
         if (routingTable.shard(params.getFollowShardId().id()).primaryShard().started() == false) {
             throw new IllegalArgumentException("Not all copies of follow shard are started");