This is an automated email from the ASF dual-hosted git repository.
petrov-mg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 029bba0b804 IGNITE-28850 Fixed propagation of finalization process ID
to joining nodes (#13321)
029bba0b804 is described below
commit 029bba0b8043951f2ede7743afe63fcbbe28fa7b
Author: Mikhail Petrov <[email protected]>
AuthorDate: Fri Jul 3 21:07:44 2026 +0300
IGNITE-28850 Fixed propagation of finalization process ID to joining nodes
(#13321)
---
.../processors/rollingupgrade/AbstractProcess.java | 4 +--
.../rollingupgrade/RollingUpgradeNodeData.java | 15 ++++++--
.../rollingupgrade/RollingUpgradeProcessor.java | 41 +++++++++++-----------
.../ClusterVersionsRollingUpgradeTest.java | 27 +++++++-------
4 files changed, 50 insertions(+), 37 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 2709b44440e..40fdee7fa93 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 @@ abstract class AbstractProcess {
}
/** */
- 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 e13b5363cc6..c484850a792 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 fe2dd268d90..cda1126ef04 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 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
RollingUpgradeNodeData gridData = data.commonData();
isVerUpgradeEnabled = gridData.isVersionUpgradeEnabled;
+ curFinalizeProcId = gridData.curFinalizeProcId;
isNodeFenceActive = gridData.isNodeFenceActive;
featureMgr.onGridDataReceived(gridData.activeFeatures);
@@ -342,7 +349,7 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
/** */
private RollingUpgradeNodeData collectRollingUpgradeNodeData() {
- return new RollingUpgradeNodeData(isVerUpgradeEnabled,
isNodeFenceActive, featureMgr.activeFeatures());
+ return new RollingUpgradeNodeData(isVerUpgradeEnabled,
curFinalizeProcId, isNodeFenceActive, featureMgr.activeFeatures());
}
/** */
@@ -407,12 +414,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
/** */
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 class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
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 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
/** */
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
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 a8606708e62..ff45becca85 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.InitMessage;
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 class ClusterVersionsRollingUpgradeTest extends
AbstractRollingUpgradeTes
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 class ClusterVersionsRollingUpgradeTest extends
AbstractRollingUpgradeTes
spi(grid(1)).waitForBlocked();
- UUID activeFinalizeProcId = extractActiveFinalizeProcessId(1);
+ UUID startedFinalizeProcId = extractFinalizeProcessId(1);
CountDownLatch finalizeCompleteStartedLatch = new CountDownLatch(1);
AtomicReference<TcpDiscoveryAbstractMessage> finalizeCompleteStartMsg
= new AtomicReference<>();
@@ -771,7 +772,7 @@ public class ClusterVersionsRollingUpgradeTest extends
AbstractRollingUpgradeTes
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 class ClusterVersionsRollingUpgradeTest extends
AbstractRollingUpgradeTes
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 @@ public class ClusterVersionsRollingUpgradeTest extends
AbstractRollingUpgradeTes
}
/** */
- 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");
}
/** */
