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/RollingUpgradeNodeData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeNodeData.java index e13b5363cc6f3..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 @@ -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; } 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..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 @@ -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 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 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 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 distinctNodeVersions = distinctClusterProductVersions(); @@ -499,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(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 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 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..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 @@ -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,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<>(); @@ -771,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(); @@ -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"); } /** */