IGNITE-28711 Introduced IgniteFeature related data structures

modules/commons/src/main/java/org/apache/ignite/lang/IgniteProductVersion.java

@@ -241,6 +241,11 @@ public int compareToIgnoreTimestamp(@NotNull IgniteProductVersion o) {
         return Integer.compare(maintenance, o.maintenance);
     }
 
+    /** @return String representation of the current version as a three-part version number. */
+    public String semanticName() {
+        return major + "." + minor + "." + maintenance;
+    }
+
     /** {@inheritDoc} */
     @Override public boolean equals(Object o) {
         if (this == o)
@@ -291,11 +296,7 @@ public int compareToIgnoreTimestamp(@NotNull IgniteProductVersion o) {
 
         hash = hash.length() > 8 ? hash.substring(0, 8) : hash;
 
-        StringBuilder sb = new StringBuilder();
-
-        sb.append(major)
-            .append(".").append(minor)
-            .append(".").append(maintenance);
+        StringBuilder sb = new StringBuilder(semanticName());
 
         if (!F.isEmpty(stage))
             sb.append("-").append(stage);

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -653,7 +653,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(GroupKeyEncrypted.class);
         withNoSchema(NodeEncryptionKeys.class);
 
-        // [13000 - 13300]: Control, configuration, diagnostincs and other messages.
+        // [13000 - 13300]: Control, configuration, diagnostics and other messages.
         msgIdx = 13000;
         withSchema(GridEventStorageMessage.class);
         withNoSchema(ChangeGlobalStateMessage.class);

modules/core/src/main/java/org/apache/ignite/internal/GridComponent.java

@@ -82,7 +82,10 @@ enum DiscoveryDataExchangeType {
         PERFORMANCE_STAT_PROC,
 
         /** Event storage manager. */
-        EVENT_MGR;
+        EVENT_MGR,
+
+        /** Rolling upgrade processor. */
+        ROLLING_UPGRADE_PROC;
 
         /** Cached values array. */
         public static final DiscoveryDataExchangeType[] VALUES = values();

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java

@@ -782,7 +782,12 @@ else if (customMsg instanceof ChangeGlobalStateMessage) {
                     discoEvtHnd.discoCache = discoCache;
 
                     if (!ctx.clientDisconnected()) {
-                        // The security processor must be notified first, since {@link IgniteSecurity#onLocalJoin}
+                        // The Rolling Upgrade Feature Manager must be notified first, as {@link IgniteFeatureManager#onLocalJoin}
+                        // completes initialization of the local node's Active Feature Set based on data received from the cluster.
+                        // The Active Feature Set, in turn, determines the node's overall behavior.
+                        ctx.rollingUpgrade().features().onLocalJoin();
+
+                        // The security processor must be notified second, since {@link IgniteSecurity#onLocalJoin}
                         // finishes local node security context initialization that can be demanded by other Ignite
                         // components.
                         ctx.security().onLocalJoin();

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java

@@ -1489,7 +1489,7 @@ public void applyReencryptionStartRecord(ReencryptionStartRecord rec) {
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> prepareMasterKeyChange(MasterKeyChangeRequest req) {
+    private IgniteInternalFuture<Message> prepareMasterKeyChange(UUID ignored, MasterKeyChangeRequest req) {
         if (masterKeyChangeRequest != null) {
             return new GridFinishedFuture<>(new IgniteException("Master key change was rejected. " +
                 "The previous change was not completed."));
@@ -1552,7 +1552,7 @@ else if (U.isLocalNodeCoordinator(ctx.discovery()))
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> performMasterKeyChange(MasterKeyChangeRequest req) {
+    private IgniteInternalFuture<Message> performMasterKeyChange(UUID ignored, MasterKeyChangeRequest req) {
         if (masterKeyChangeRequest == null || !masterKeyChangeRequest.equals(req))
             return new GridFinishedFuture<>(new IgniteException("Unknown master key change was rejected."));
 

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java

@@ -183,7 +183,7 @@ public IgniteFuture<Void> start(Collection<String> cacheOrGrpNames) {
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> prepare(ChangeCacheEncryptionRequest req) {
+    private IgniteInternalFuture<Message> prepare(UUID ignored, ChangeCacheEncryptionRequest req) {
         if (ctx.clientNode())
             return new GridFinishedFuture<>();
 
@@ -285,7 +285,7 @@ else if (U.isLocalNodeCoordinator(ctx.discovery()))
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> perform(ChangeCacheEncryptionRequest req) {
+    private IgniteInternalFuture<Message> perform(UUID ignored, ChangeCacheEncryptionRequest req) {
         if (this.req == null || !this.req.equals(req))
             return new GridFinishedFuture<>(new IgniteException("Unknown cache group key change was rejected."));
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java

@@ -765,7 +765,7 @@ private void deleteDirectory(File dir) throws IOException {
      * @param req Request on snapshot creation.
      * @return Future which will be completed when a snapshot has been started.
      */
-    private IgniteInternalFuture<SnapshotOperationResponse> initLocalSnapshotStartStage(SnapshotOperationRequest req) {
+    private IgniteInternalFuture<SnapshotOperationResponse> initLocalSnapshotStartStage(UUID ignored, SnapshotOperationRequest req) {
         // Executed inside discovery notifier thread, prior to firing discovery custom event,
         // so it is safe to set new snapshot task inside this method without synchronization.
         if (curSnpOp != null) {
@@ -1253,7 +1253,7 @@ private IgniteInternalFuture<Void> completeHandlersAsyncIfNeeded(
      * @param endReq Snapshot creation end request.
      * @return Future which will be completed when the snapshot will be finalized.
      */
-    private IgniteInternalFuture<SnapshotOperationResponse> initLocalSnapshotEndStage(SnapshotOperationEndRequest endReq) {
+    private IgniteInternalFuture<SnapshotOperationResponse> initLocalSnapshotEndStage(UUID ignored, SnapshotOperationEndRequest endReq) {
         SnapshotOperation snpOp = curSnpOp;
 
         if (snpOp == null || !Objects.equals(endReq.requestId(), snpOp.request().requestId()))

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java

@@ -311,7 +311,7 @@ private void reducePartitionsHashesResults(
     }
 
     /** Phase 2 beginning.  */
-    private IgniteInternalFuture<SnapshotCheckResponse> validateParts(SnapshotCheckProcessRequest req) {
+    private IgniteInternalFuture<SnapshotCheckResponse> validateParts(UUID ignored, SnapshotCheckProcessRequest req) {
         if (!req.nodes().contains(kctx.localNodeId()))
             return new GridFinishedFuture<>();
 
@@ -477,7 +477,7 @@ private static Exception asException(Throwable th) {
     }
 
     /** Phase 1 beginning: prepare, collect and check local metas. */
-    private IgniteInternalFuture<SnapshotCheckResponse> prepareAndCheckMetas(SnapshotCheckProcessRequest req) {
+    private IgniteInternalFuture<SnapshotCheckResponse> prepareAndCheckMetas(UUID ignored, SnapshotCheckProcessRequest req) {
         if (!req.nodes().contains(kctx.localNodeId()) && clusterOpFuts.get(req.requestId()) == null)
             return new GridFinishedFuture<>();
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java

@@ -629,7 +629,7 @@ private void ensureCacheAbsent(String name) {
      * @param req Request to prepare cache group restore from the snapshot.
      * @return Result future.
      */
-    private IgniteInternalFuture<SnapshotRestoreOperationResponse> prepare(SnapshotOperationRequest req) {
+    private IgniteInternalFuture<SnapshotRestoreOperationResponse> prepare(UUID ignored, SnapshotOperationRequest req) {
         if (ctx.clientNode())
             return new GridFinishedFuture<>();
 
@@ -912,7 +912,7 @@ private static GridAffinityAssignmentCache calculateAffinity(
      * @param req Request.
      * @return Future which will be completed when the preload ends.
      */
-    private IgniteInternalFuture<Message> preload(SnapshotRestoreStartRequest req) {
+    private IgniteInternalFuture<Message> preload(UUID ignored, SnapshotRestoreStartRequest req) {
         if (ctx.clientNode())
             return new GridFinishedFuture<>();
 
@@ -1263,7 +1263,7 @@ private void finishPreload(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwab
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> cacheStart(SnapshotRestoreStartRequest req) {
+    private IgniteInternalFuture<Message> cacheStart(UUID ignored, SnapshotRestoreStartRequest req) {
         if (ctx.clientNode())
             return new GridFinishedFuture<>();
 
@@ -1333,7 +1333,7 @@ private void finishCacheStart(UUID reqId, Map<UUID, Message> res, Map<UUID, Thro
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> cacheStop(SnapshotRestoreStartRequest req) {
+    private IgniteInternalFuture<Message> cacheStop(UUID ignored, SnapshotRestoreStartRequest req) {
         if (!U.isLocalNodeCoordinator(ctx.discovery()))
             return new GridFinishedFuture<>();
 
@@ -1384,7 +1384,7 @@ private void finishCacheStop(UUID reqId, Map<UUID, Message> res, Map<UUID, Throw
      * @param req Request ID.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> incrementalSnapshotRestore(Message req) {
+    private IgniteInternalFuture<Message> incrementalSnapshotRestore(UUID ignored, Message req) {
         SnapshotRestoreContext opCtx0 = opCtx;
 
         if (ctx.clientNode() || opCtx0 == null || !opCtx0.nodes().contains(ctx.localNodeId()))
@@ -1652,7 +1652,7 @@ private static ClusterTopologyCheckedException checkNodeLeft(Collection<UUID> re
      * @param req Request.
      * @return Result future.
      */
-    private IgniteInternalFuture<Message> rollback(SnapshotRestoreStartRequest req) {
+    private IgniteInternalFuture<Message> rollback(UUID ignored, SnapshotRestoreStartRequest req) {
         if (ctx.clientNode())
             return new GridFinishedFuture<>();
 

modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsProcessor.java

@@ -124,7 +124,7 @@ public PerformanceStatisticsProcessor(GridKernalContext ctx) {
             });
 
         rotateProc = new DistributedProcess<>(ctx, PERFORMANCE_STATISTICS_ROTATE,
-            req -> ctx.closure().callLocalSafe(() -> {
+            (id, req) -> ctx.closure().callLocalSafe(() -> {
                 rotateWriter();
 
                 return null;

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

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.rollingupgrade;
+
+import java.util.UUID;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+abstract class AbstractProcess {
+    /** */
+    @Nullable private UUID locInitOpId;
+
+    /** */
+    @Nullable private GridFutureAdapter<Void> locInitOpFut;
+
+    /** */
+    protected abstract UUID startInternal();
+
+    /** */
+    protected synchronized IgniteInternalFuture<Void> start() {
+        if (locInitOpFut != null)
+            return locInitOpFut;
+
+        locInitOpId = startInternal();
+        locInitOpFut = new GridFutureAdapter<>();
+
+        return locInitOpFut;
+    }
+
+    /** */
+    protected synchronized void finishProcess(UUID reqId, @Nullable Throwable err) {
+        if (!reqId.equals(locInitOpId))
+            return;
+
+        locInitOpFut.onDone(err);
+
+        locInitOpId = null;
+        locInitOpFut = null;
+    }
+
+    /** */
+    protected synchronized void abort(String reasonMsg) {
+        locInitOpId = null;
+
+        if (locInitOpFut != null) {
+            locInitOpFut.onDone(new IgniteException("Operation was aborted [reason=" + reasonMsg + ']'));
+
+            locInitOpFut = null;
+        }
+    }
+}

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

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.rollingupgrade;
+
+import java.io.Serializable;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteProductFeatures;
+
+/** */
+public class RollingUpgradeNodeData implements Serializable {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** */
+    private final boolean isVersionUpgradeEnabled;
+
+    /** */
+    private final IgniteProductFeatures activeFeatures;
+
+    /** */
+    private final boolean isNodeFenceActive;
+
+    /** */
+    public RollingUpgradeNodeData(boolean isVersionUpgradeEnabled, boolean isNodeFenceActive, IgniteProductFeatures activeFeatures) {
+        this.isVersionUpgradeEnabled = isVersionUpgradeEnabled;
+        this.isNodeFenceActive = isNodeFenceActive;
+        this.activeFeatures = activeFeatures;
+    }
+
+    /** */
+    public boolean isVersionUpgradeEnabled() {
+        return isVersionUpgradeEnabled;
+    }
+
+    /** */
+    public boolean isNodeFenceActive() {
+        return isNodeFenceActive;
+    }
+
+    /** */
+    public IgniteProductFeatures activeFeatures() {
+        return activeFeatures;
+    }
+}