avoid hotspot in a mixed cluster

Author: dnhatn

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

@@ -78,7 +78,6 @@
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
-import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.ccr.CcrLicenseChecker.wrapClient;
@@ -122,17 +121,14 @@ public void validate(ShardFollowTask params, ClusterState clusterState) {
 
     @Override
     public Assignment getAssignment(final ShardFollowTask params, final ClusterState clusterState) {
-        DiscoveryNode node = selectLeastLoadedNode(
+        final DiscoveryNode selectedNode = selectLeastLoadedNode(
             clusterState,
-            ((Predicate<DiscoveryNode>) DiscoveryNode::isDataNode).and(DiscoveryNode::isRemoteClusterClient)
+            node -> node.isDataNode() && (node.isRemoteClusterClient() || node.getVersion().before(DiscoveryNode.PLUGGABLE_ROLES_VERSION))
         );
-        if (node == null) {
-            node = selectLeastLoadedNode(clusterState, n -> n.isDataNode() && n.getVersion().before(DiscoveryNode.PLUGGABLE_ROLES_VERSION));
-        }
-        if (node == null) {
+        if (selectedNode == null) {
             return NO_ASSIGNMENT;
         } else {
-            return new Assignment(node.getId(), "node is the least loaded data node and remote cluster client");
+            return new Assignment(selectedNode.getId(), "node is the least loaded data node and remote cluster client");
         }
     }