IGNITE-28850 Fixed propagation of finalization process ID to joining nodes

modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractProcess.java

@@ -57,11 +57,11 @@ protected synchronized void finishProcess(UUID reqId, @Nullable Throwable err) {
     }
 
     /** */
-    protected synchronized void abort(String reasonMsg) {
+    protected synchronized void abort(String reason) {
         locInitOpId = null;
 
         if (locInitOpFut != null) {
-            locInitOpFut.onDone(new IgniteException("Operation was aborted [reason=" + reasonMsg + ']'));
+            locInitOpFut.onDone(new IgniteException("Operation was aborted [reason=" + reason + ']'));
 
             locInitOpFut = null;
         }

modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.rollingupgrade;
 
+import java.util.UUID;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteProductFeatures;
 import org.apache.ignite.plugin.extensions.communication.Message;
@@ -29,20 +30,30 @@ public class RollingUpgradeNodeData implements Message {
 
     /** */
     @Order(1)
-    IgniteProductFeatures activeFeatures;
+    UUID curFinalizeProcId;
 
     /** */
     @Order(2)
     boolean isNodeFenceActive;
 
+    /** */
+    @Order(3)
+    IgniteProductFeatures activeFeatures;
+
     /** */
     public RollingUpgradeNodeData() {
         // No-op.
     }
 
     /** */
-    public RollingUpgradeNodeData(boolean isVersionUpgradeEnabled, boolean isNodeFenceActive, IgniteProductFeatures activeFeatures) {
+    public RollingUpgradeNodeData(
+        boolean isVersionUpgradeEnabled,
+        UUID curFinalizeProcId,
+        boolean isNodeFenceActive,
+        IgniteProductFeatures activeFeatures
+    ) {
         this.isVersionUpgradeEnabled = isVersionUpgradeEnabled;
+        this.curFinalizeProcId = curFinalizeProcId;
         this.isNodeFenceActive = isNodeFenceActive;
         this.activeFeatures = activeFeatures;
     }

modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java

@@ -88,6 +88,12 @@ public class RollingUpgradeProcessor extends GridProcessorAdapter implements Dis
     /** */
     private volatile boolean isVerUpgradeEnabled;
 
+    /**
+     * We rely on the guarantee that this varable is only accessed from the Discovery thread.
+     * Therefore, synchronization is not required.
+     */
+    @Nullable private volatile UUID curFinalizeProcId;
+
     /** */
     public RollingUpgradeProcessor(GridKernalContext ctx) {
         this(
@@ -216,6 +222,7 @@ RollingUpgradeState state() {
         RollingUpgradeNodeData gridData = data.commonData();
 
         isVerUpgradeEnabled = gridData.isVersionUpgradeEnabled;
+        curFinalizeProcId = gridData.curFinalizeProcId;
         isNodeFenceActive = gridData.isNodeFenceActive;
 
         featureMgr.onGridDataReceived(gridData.activeFeatures);
@@ -342,7 +349,7 @@ private SortedSet<IgniteProductVersion> distinctClusterProductVersions() {
 
     /** */
     private RollingUpgradeNodeData collectRollingUpgradeNodeData() {
-        return new RollingUpgradeNodeData(isVerUpgradeEnabled, isNodeFenceActive, featureMgr.activeFeatures());
+        return new RollingUpgradeNodeData(isVerUpgradeEnabled, curFinalizeProcId, isNodeFenceActive, featureMgr.activeFeatures());
     }
 
     /** */
@@ -407,12 +414,6 @@ private class ClusterVersionFinalizationProcess extends AbstractProcess {
         /** */
         private final DistributedProcess<Message, Message> completePhase;
 
-        /**
-         * We rely on the guarantee that {@code activeProcId} is only accessed from the Discovery thread.
-         * Therefore, synchronization is not required.
-         */
-        @Nullable private volatile UUID activeProcId;
-
         /** */
         public ClusterVersionFinalizationProcess() {
             preparePhase = new DistributedProcess<>(
@@ -447,37 +448,37 @@ public ClusterVersionFinalizationProcess() {
             if (err != null)
                 U.error(log, "Cluster version finalization process failed [procId=" + reqId + ']', err);
 
-            if (reqId.equals(activeProcId))
-                activeProcId = null;
+            if (reqId.equals(curFinalizeProcId))
+                curFinalizeProcId = null;
 
             super.finishProcess(reqId, err);
         }
 
         /** {@inheritDoc} */
-        @Override protected void abort(String reasonMsg) {
-            U.warn(log, "Cluster version finalization process has been aborted [procId=" + activeProcId + ", reason=" + reasonMsg + ']');
+        @Override protected void abort(String reason) {
+            U.warn(log, "Cluster version finalization process has been aborted [procId=" + curFinalizeProcId + ", reason=" + reason + ']');
 
-            activeProcId = null;
+            curFinalizeProcId = null;
             isNodeFenceActive = false;
 
-            super.abort(reasonMsg);
+            super.abort(reason);
         }
 
         /** */
         boolean isInProgress() {
-            return isVerUpgradeEnabled && activeProcId != null;
+            return isVerUpgradeEnabled && curFinalizeProcId != null;
         }
 
         /** */
         private IgniteInternalFuture<Message> executePreparePhase(UUID reqId, Message req) {
-            if (activeProcId != null) {
+            if (curFinalizeProcId != null) {
                 U.error(log, "Failed to handle cluster version finalization request. Another process is " +
-                    "already in progress [curProcId=" + reqId + ", activeProcId=" + activeProcId + ']');
+                    "already in progress [curProcId=" + reqId + ", activeProcId=" + curFinalizeProcId + ']');
 
                 return new GridFinishedFuture<>(new IgniteException("Cluster version finalization process is already in progress"));
             }
 
-            activeProcId = reqId;
+            curFinalizeProcId = reqId;
 
             synchronized (topGuard) {
                 Set<IgniteProductVersion> distinctNodeVersions = distinctClusterProductVersions();
@@ -499,18 +500,18 @@ private IgniteInternalFuture<Message> executePreparePhase(UUID reqId, Message re
         /** */
         private void finishPreparePhase(UUID reqId, Map<UUID, Message> responses, Map<UUID, Throwable> errors) {
             if (!F.isEmpty(errors)) {
-                if (reqId.equals(activeProcId))
+                if (reqId.equals(curFinalizeProcId))
                     isNodeFenceActive = false;
 
                 finishProcess(reqId, firstError(errors));
             }
-            else if (reqId.equals(activeProcId) && U.isLocalNodeCoordinator(ctx.discovery()))
+            else if (reqId.equals(curFinalizeProcId) && U.isLocalNodeCoordinator(ctx.discovery()))
                 completePhase.start(reqId, null);
         }
 
         /** */
         private IgniteInternalFuture<Message> executeCompletePhase(UUID reqId, Message req) {
-            if (!reqId.equals(activeProcId)) {
+            if (!reqId.equals(curFinalizeProcId)) {
                 // This condition is guaranteed to occur only when cluster version finalization is aborted
                 // after the prepare phase has completed successfully but before the completion phase begins.
                 // Aborting cluster version finalization is mutually exclusive with the finalization completion

modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java

@@ -41,6 +41,7 @@
 import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
 import org.apache.ignite.internal.util.lang.ConsumerX;
 import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.X;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.spi.IgniteSpiException;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
@@ -547,13 +548,13 @@ public void testConcurrentFinalizationErrorPreserveNodeFence() throws Exception
 
             waitForBlockedDiscoveryMessages(grid(0), 2, InitMessage.class);
 
-            UUID activeFinalizeProcId = extractActiveFinalizeProcessId(1);
+            UUID startedFinalizeProcId = extractFinalizeProcessId(1);
 
             spi(grid(2)).blockMessages((node, msg) -> {
                 if (!(msg instanceof SingleNodeMessage<?> singleNodeMsg))
                     return false;
 
-                return singleNodeMsg.processId().equals(activeFinalizeProcId);
+                return singleNodeMsg.processId().equals(startedFinalizeProcId);
             });
 
             discoveryRingMessageWorkerQueue(grid(0)).unblock();
@@ -763,15 +764,15 @@ public void testAbortBeforeFinalizationComplete() throws Exception {
 
         spi(grid(1)).waitForBlocked();
 
-        UUID activeFinalizeProcId = extractActiveFinalizeProcessId(1);
+        UUID startedFinalizeProcId = extractFinalizeProcessId(1);
 
         CountDownLatch finalizeCompleteStartedLatch = new CountDownLatch(1);
         AtomicReference<TcpDiscoveryAbstractMessage> finalizeCompleteStartMsg = new AtomicReference<>();
 
         discoveryRingMessageWorkerQueue(grid(0)).startMessageIntercepting(m -> {
             if ((m instanceof TcpDiscoveryCustomEventMessage customMsg)
                 && (customMsg.message() instanceof InitMessage<?> initMsg)
-                && initMsg.processId().equals(activeFinalizeProcId)
+                && initMsg.processId().equals(startedFinalizeProcId)
             ) {
                 finalizeCompleteStartMsg.set(m);
                 finalizeCompleteStartedLatch.countDown();
@@ -935,12 +936,12 @@ public void testConcurrentFinalizationAndNodeJoin() throws Exception {
 
         finalizeFut.get(getTestTimeout(), MILLISECONDS);
 
-        GridTestUtils.assertThrowsAnyCause(
-            log,
-            () -> startFut.get(getTestTimeout(), MILLISECONDS),
-            IgniteSpiException.class,
-            "Node joins are not allowed during cluster version finalization"
-        );
+        try {
+            startFut.get(getTestTimeout(), MILLISECONDS);
+        }
+        catch (IgniteCheckedException e) {
+            assertTrue(X.hasCause(e, "Node joins are not allowed during cluster version finalization", IgniteSpiException.class));
+        }
     }
 
     /** */
@@ -981,10 +982,10 @@ private void startCluster(String ver) throws Exception {
     }
 
     /** */
-    private UUID extractActiveFinalizeProcessId(int nodeIdx) {
-        Object enable = U.field(grid(nodeIdx).context().rollingUpgrade(), "finalizeProc");
+    private UUID extractFinalizeProcessId(int nodeIdx) {
+        Object proc = U.field(grid(nodeIdx).context().rollingUpgrade(), "finalizeProc");
 
-        return U.field(enable, "locInitOpId");
+        return U.field(proc, "locInitOpId");
     }
 
     /** */