From b14960e5e8055f1ada483febf6db3fc499f64d57 Mon Sep 17 00:00:00 2001 From: Mikhail Petrov Date: Fri, 3 Jul 2026 10:15:00 +0300 Subject: [PATCH 1/4] IGNITE-28850 Fixed propagation of finalization process ID to joining nodes --- .../RollingUpgradeNodeData.java | 15 +++++++- .../RollingUpgradeProcessor.java | 38 ++++++++++--------- .../ClusterVersionsRollingUpgradeTest.java | 13 ++++--- 3 files changed, 40 insertions(+), 26 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java index e13b5363cc6f3..38c543a08da16 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java +++ b/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 activeFinalizeProcId; /** */ @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 activeFinalizeProcId, + boolean isNodeFenceActive, + IgniteProductFeatures activeFeatures + ) { this.isVersionUpgradeEnabled = isVersionUpgradeEnabled; + this.activeFinalizeProcId = activeFinalizeProcId; this.isNodeFenceActive = isNodeFenceActive; this.activeFeatures = activeFeatures; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java index fe2dd268d9031..0d9066808a28b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java +++ b/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 activeFinalizeProcId; + /** */ public RollingUpgradeProcessor(GridKernalContext ctx) { this( @@ -216,6 +222,7 @@ RollingUpgradeState state() { RollingUpgradeNodeData gridData = data.commonData(); isVerUpgradeEnabled = gridData.isVersionUpgradeEnabled; + activeFinalizeProcId = gridData.activeFinalizeProcId; isNodeFenceActive = gridData.isNodeFenceActive; featureMgr.onGridDataReceived(gridData.activeFeatures); @@ -342,7 +349,7 @@ private SortedSet distinctClusterProductVersions() { /** */ private RollingUpgradeNodeData collectRollingUpgradeNodeData() { - return new RollingUpgradeNodeData(isVerUpgradeEnabled, isNodeFenceActive, featureMgr.activeFeatures()); + return new RollingUpgradeNodeData(isVerUpgradeEnabled, activeFinalizeProcId, isNodeFenceActive, featureMgr.activeFeatures()); } /** */ @@ -407,12 +414,6 @@ private class ClusterVersionFinalizationProcess extends AbstractProcess { /** */ private final DistributedProcess 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,17 +448,18 @@ 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(activeFinalizeProcId)) + activeFinalizeProcId = 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 + ']'); + U.warn(log, + "Cluster version finalization process has been aborted [procId=" + activeFinalizeProcId + ", reason=" + reasonMsg + ']'); - activeProcId = null; + activeFinalizeProcId = null; isNodeFenceActive = false; super.abort(reasonMsg); @@ -465,19 +467,19 @@ public ClusterVersionFinalizationProcess() { /** */ boolean isInProgress() { - return isVerUpgradeEnabled && activeProcId != null; + return isVerUpgradeEnabled && activeFinalizeProcId != null; } /** */ private IgniteInternalFuture executePreparePhase(UUID reqId, Message req) { - if (activeProcId != null) { + if (activeFinalizeProcId != 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=" + activeFinalizeProcId + ']'); return new GridFinishedFuture<>(new IgniteException("Cluster version finalization process is already in progress")); } - activeProcId = reqId; + activeFinalizeProcId = reqId; synchronized (topGuard) { Set distinctNodeVersions = distinctClusterProductVersions(); @@ -499,18 +501,18 @@ private IgniteInternalFuture executePreparePhase(UUID reqId, Message re /** */ private void finishPreparePhase(UUID reqId, Map responses, Map errors) { if (!F.isEmpty(errors)) { - if (reqId.equals(activeProcId)) + if (reqId.equals(activeFinalizeProcId)) isNodeFenceActive = false; finishProcess(reqId, firstError(errors)); } - else if (reqId.equals(activeProcId) && U.isLocalNodeCoordinator(ctx.discovery())) + else if (reqId.equals(activeFinalizeProcId) && U.isLocalNodeCoordinator(ctx.discovery())) completePhase.start(reqId, null); } /** */ private IgniteInternalFuture executeCompletePhase(UUID reqId, Message req) { - if (!reqId.equals(activeProcId)) { + if (!reqId.equals(activeFinalizeProcId)) { // 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 diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java index a8606708e6238..8aa8c9ec26584 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java +++ b/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; @@ -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)); + } } /** */ From 6672521e9698f0c1452d474432154c31a597c039 Mon Sep 17 00:00:00 2001 From: Mikhail Petrov Date: Fri, 3 Jul 2026 10:40:20 +0300 Subject: [PATCH 2/4] IGNITE-28850 Fixed propagation of finalization process ID to joining nodes --- .../rollingupgrade/AbstractProcess.java | 4 +-- .../RollingUpgradeProcessor.java | 33 +++++++++---------- 2 files changed, 18 insertions(+), 19 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractProcess.java index 2709b44440e4d..40fdee7fa9326 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractProcess.java +++ b/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; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java index 0d9066808a28b..3fc4d46a4fe30 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java @@ -92,7 +92,7 @@ public class RollingUpgradeProcessor extends GridProcessorAdapter implements Dis * We rely on the guarantee that this varable is only accessed from the Discovery thread. * Therefore, synchronization is not required. */ - @Nullable private volatile UUID activeFinalizeProcId; + @Nullable private volatile UUID curFinalizeProcId; /** */ public RollingUpgradeProcessor(GridKernalContext ctx) { @@ -222,7 +222,7 @@ RollingUpgradeState state() { RollingUpgradeNodeData gridData = data.commonData(); isVerUpgradeEnabled = gridData.isVersionUpgradeEnabled; - activeFinalizeProcId = gridData.activeFinalizeProcId; + curFinalizeProcId = gridData.activeFinalizeProcId; isNodeFenceActive = gridData.isNodeFenceActive; featureMgr.onGridDataReceived(gridData.activeFeatures); @@ -349,7 +349,7 @@ private SortedSet distinctClusterProductVersions() { /** */ private RollingUpgradeNodeData collectRollingUpgradeNodeData() { - return new RollingUpgradeNodeData(isVerUpgradeEnabled, activeFinalizeProcId, isNodeFenceActive, featureMgr.activeFeatures()); + return new RollingUpgradeNodeData(isVerUpgradeEnabled, curFinalizeProcId, isNodeFenceActive, featureMgr.activeFeatures()); } /** */ @@ -448,38 +448,37 @@ public ClusterVersionFinalizationProcess() { if (err != null) U.error(log, "Cluster version finalization process failed [procId=" + reqId + ']', err); - if (reqId.equals(activeFinalizeProcId)) - activeFinalizeProcId = 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=" + activeFinalizeProcId + ", reason=" + reasonMsg + ']'); + @Override protected void abort(String reason) { + U.warn(log, "Cluster version finalization process has been aborted [procId=" + curFinalizeProcId + ", reason=" + reason + ']'); - activeFinalizeProcId = null; + curFinalizeProcId = null; isNodeFenceActive = false; - super.abort(reasonMsg); + super.abort(reason); } /** */ boolean isInProgress() { - return isVerUpgradeEnabled && activeFinalizeProcId != null; + return isVerUpgradeEnabled && curFinalizeProcId != null; } /** */ private IgniteInternalFuture executePreparePhase(UUID reqId, Message req) { - if (activeFinalizeProcId != null) { + if (curFinalizeProcId != null) { U.error(log, "Failed to handle cluster version finalization request. Another process is " + - "already in progress [curProcId=" + reqId + ", activeProcId=" + activeFinalizeProcId + ']'); + "already in progress [curProcId=" + reqId + ", activeProcId=" + curFinalizeProcId + ']'); return new GridFinishedFuture<>(new IgniteException("Cluster version finalization process is already in progress")); } - activeFinalizeProcId = reqId; + curFinalizeProcId = reqId; synchronized (topGuard) { Set distinctNodeVersions = distinctClusterProductVersions(); @@ -501,18 +500,18 @@ private IgniteInternalFuture executePreparePhase(UUID reqId, Message re /** */ private void finishPreparePhase(UUID reqId, Map responses, Map errors) { if (!F.isEmpty(errors)) { - if (reqId.equals(activeFinalizeProcId)) + if (reqId.equals(curFinalizeProcId)) isNodeFenceActive = false; finishProcess(reqId, firstError(errors)); } - else if (reqId.equals(activeFinalizeProcId) && U.isLocalNodeCoordinator(ctx.discovery())) + else if (reqId.equals(curFinalizeProcId) && U.isLocalNodeCoordinator(ctx.discovery())) completePhase.start(reqId, null); } /** */ private IgniteInternalFuture executeCompletePhase(UUID reqId, Message req) { - if (!reqId.equals(activeFinalizeProcId)) { + 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 From fdc0b16e321fe0f097f5d449ef9477496550cd35 Mon Sep 17 00:00:00 2001 From: Mikhail Petrov Date: Fri, 3 Jul 2026 15:09:49 +0300 Subject: [PATCH 3/4] IGNITE-28850 Fixed propagation of finalization process ID to joining nodes --- .../processors/rollingupgrade/RollingUpgradeNodeData.java | 6 +++--- .../processors/rollingupgrade/RollingUpgradeProcessor.java | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java index 38c543a08da16..c484850a79281 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java @@ -30,7 +30,7 @@ public class RollingUpgradeNodeData implements Message { /** */ @Order(1) - UUID activeFinalizeProcId; + UUID curFinalizeProcId; /** */ @Order(2) @@ -48,12 +48,12 @@ public RollingUpgradeNodeData() { /** */ public RollingUpgradeNodeData( boolean isVersionUpgradeEnabled, - UUID activeFinalizeProcId, + UUID curFinalizeProcId, boolean isNodeFenceActive, IgniteProductFeatures activeFeatures ) { this.isVersionUpgradeEnabled = isVersionUpgradeEnabled; - this.activeFinalizeProcId = activeFinalizeProcId; + this.curFinalizeProcId = curFinalizeProcId; this.isNodeFenceActive = isNodeFenceActive; this.activeFeatures = activeFeatures; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java index 3fc4d46a4fe30..cda1126ef0497 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java @@ -222,7 +222,7 @@ RollingUpgradeState state() { RollingUpgradeNodeData gridData = data.commonData(); isVerUpgradeEnabled = gridData.isVersionUpgradeEnabled; - curFinalizeProcId = gridData.activeFinalizeProcId; + curFinalizeProcId = gridData.curFinalizeProcId; isNodeFenceActive = gridData.isNodeFenceActive; featureMgr.onGridDataReceived(gridData.activeFeatures); From e5c2e500185b055c40ec7b9f71a924adfd642313 Mon Sep 17 00:00:00 2001 From: Mikhail Petrov Date: Fri, 3 Jul 2026 15:11:53 +0300 Subject: [PATCH 4/4] IGNITE-28850 Fixed propagation of finalization process ID to joining nodes --- .../ClusterVersionsRollingUpgradeTest.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java index 8aa8c9ec26584..ff45becca85b0 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/ClusterVersionsRollingUpgradeTest.java @@ -548,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(); @@ -764,7 +764,7 @@ public void testAbortBeforeFinalizationComplete() throws Exception { spi(grid(1)).waitForBlocked(); - UUID activeFinalizeProcId = extractActiveFinalizeProcessId(1); + UUID startedFinalizeProcId = extractFinalizeProcessId(1); CountDownLatch finalizeCompleteStartedLatch = new CountDownLatch(1); AtomicReference finalizeCompleteStartMsg = new AtomicReference<>(); @@ -772,7 +772,7 @@ public void testAbortBeforeFinalizationComplete() throws Exception { 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(); @@ -982,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"); } /** */