Extend version barrier to all upgrades (#73358)

Today when upgrading to the next major version we have a so-called
_major version barrier_: once the cluster comprises nodes of the new
major version then nodes of the previous major version are prevented
from joining the cluster. This means we can be certain that
`clusterState.nodes().getMinNodeVersion().major` will never decrease, so
we can implement upgrade logic that relies on the cluster remaining in
its wholly-upgraded state.

This commit generalises this behaviour to apply to all upgrades, so that
we can be certain that `clusterState.nodes().getMinNodeVersion()` will
never decrease in a running cluster.

Closes #72911

docs/reference/upgrade/rolling_upgrade.asciidoc

@@ -220,7 +220,7 @@ any new functionality is disabled or operates in a backward compatible mode
 until all nodes in the cluster are upgraded. New functionality becomes
 operational once the upgrade is complete and all nodes are running the new
 version. Once that has happened, there's no way to return to operating in a
-backward compatible mode. Nodes running the previous major version will not be
+backward compatible mode. Nodes running the previous version will not be
 allowed to join the fully-updated cluster.
 
 In the unlikely case of a network malfunction during the upgrade process that

server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

@@ -453,7 +453,8 @@ public class Coordinator extends AbstractLifecycleComponent implements Discovery
                 if (stateForJoinValidation.getBlocks().hasGlobalBlock(STATE_NOT_RECOVERED_BLOCK) == false) {
                     // we do this in a couple of places including the cluster update thread. This one here is really just best effort
                     // to ensure we fail as fast as possible.
-                    JoinTaskExecutor.ensureMajorVersionBarrier(joinRequest.getSourceNode().getVersion(),
+                    JoinTaskExecutor.ensureVersionBarrier(
+                        joinRequest.getSourceNode().getVersion(),
                         stateForJoinValidation.getNodes().getMinNodeVersion());
                 }
                 sendValidateJoinRequest(stateForJoinValidation, joinRequest, joinCallback);

server/src/main/java/org/elasticsearch/cluster/coordination/JoinTaskExecutor.java

@@ -121,8 +121,8 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut
 
         Version minClusterNodeVersion = newState.nodes().getMinNodeVersion();
         Version maxClusterNodeVersion = newState.nodes().getMaxNodeVersion();
-        // we only enforce major version transitions on a fully formed clusters
-        final boolean enforceMajorVersion = currentState.getBlocks().hasGlobalBlock(STATE_NOT_RECOVERED_BLOCK) == false;
+        // if the cluster is not fully-formed then the min version is not meaningful
+        final boolean enforceVersionBarrier = currentState.getBlocks().hasGlobalBlock(STATE_NOT_RECOVERED_BLOCK) == false;
         // processing any joins
         Map<String, String> joiniedNodeNameIds = new HashMap<>();
         for (final Task joinTask : joiningNodes) {
@@ -133,8 +133,8 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut
             } else {
                 final DiscoveryNode node = joinTask.node();
                 try {
-                    if (enforceMajorVersion) {
-                        ensureMajorVersionBarrier(node.getVersion(), minClusterNodeVersion);
+                    if (enforceVersionBarrier) {
+                        ensureVersionBarrier(node.getVersion(), minClusterNodeVersion);
                     }
                     ensureNodesCompatibility(node.getVersion(), minClusterNodeVersion, maxClusterNodeVersion);
                     // we do this validation quite late to prevent race conditions between nodes joining and importing dangling indices
@@ -185,7 +185,12 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut
                 }
             }
 
-            return results.build(allocationService.adaptAutoExpandReplicas(newState.nodes(nodesBuilder).build()));
+            final ClusterState updatedState = allocationService.adaptAutoExpandReplicas(newState.nodes(nodesBuilder).build());
+            assert enforceVersionBarrier == false
+                || updatedState.nodes().getMinNodeVersion().onOrAfter(currentState.nodes().getMinNodeVersion())
+                : "min node version decreased from [" + currentState.nodes().getMinNodeVersion() + "] to ["
+                + updatedState.nodes().getMinNodeVersion() + "]";
+            return results.build(updatedState);
         } else {
             // we must return a new cluster state instance to force publishing. This is important
             // for the joining node to finalize its join and set us as a master
@@ -293,15 +298,14 @@ public class JoinTaskExecutor implements ClusterStateTaskExecutor<JoinTaskExecut
     }
 
     /**
-     * ensures that the joining node's major version is equal or higher to the minClusterNodeVersion. This is needed
-     * to ensure that if the master is already fully operating under the new major version, it doesn't go back to mixed
+     * ensures that the joining node's version is equal or higher to the minClusterNodeVersion. This is needed
+     * to ensure that if the master is already fully operating under the new version, it doesn't go back to mixed
      * version mode
      **/
-    public static void ensureMajorVersionBarrier(Version joiningNodeVersion, Version minClusterNodeVersion) {
-        final byte clusterMajor = minClusterNodeVersion.major;
-        if (joiningNodeVersion.major < clusterMajor) {
-            throw new IllegalStateException("node version [" + joiningNodeVersion + "] is not supported. " +
-                "All nodes in the cluster are of a higher major [" + clusterMajor + "].");
+    public static void ensureVersionBarrier(Version joiningNodeVersion, Version minClusterNodeVersion) {
+        if (joiningNodeVersion.before(minClusterNodeVersion)) {
+            throw new IllegalStateException("node version [" + joiningNodeVersion +
+                "] may not join a cluster comprising only nodes of version [" + minClusterNodeVersion + "] or greater");
         }
     }
 

server/src/test/java/org/elasticsearch/cluster/coordination/JoinTaskExecutorTests.java

@@ -27,6 +27,7 @@ import java.util.List;
 
 import static org.elasticsearch.test.VersionUtils.maxCompatibleVersion;
 import static org.elasticsearch.test.VersionUtils.randomCompatibleVersion;
+import static org.elasticsearch.test.VersionUtils.randomVersion;
 import static org.elasticsearch.test.VersionUtils.randomVersionBetween;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -85,8 +86,9 @@ public class JoinTaskExecutorTests extends ESTestCase {
             }
         });
 
-        Version oldMajor = minNodeVersion.minimumCompatibilityVersion();
-        expectThrows(IllegalStateException.class, () -> JoinTaskExecutor.ensureMajorVersionBarrier(oldMajor, minNodeVersion));
+        final Version oldVersion = randomValueOtherThanMany(v -> v.onOrAfter(minNodeVersion),
+            () -> rarely() ? Version.fromId(minNodeVersion.id - 1) : randomVersion(random()));
+        expectThrows(IllegalStateException.class, () -> JoinTaskExecutor.ensureVersionBarrier(oldVersion, minNodeVersion));
 
         final Version minGoodVersion = maxNodeVersion.major == minNodeVersion.major ?
             // we have to stick with the same major