Ignite 13805

modules/core/src/main/java/org/apache/ignite/IgniteSnapshot.java

@@ -17,7 +17,9 @@
 
 package org.apache.ignite;
 
+import java.util.Collection;
 import org.apache.ignite.lang.IgniteFuture;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * This interface provides functionality for creating cluster-wide cache data snapshots.
@@ -48,4 +50,16 @@ public interface IgniteSnapshot {
      * @return Future which will be completed when cancel operation finished.
      */
     public IgniteFuture<Void> cancelSnapshot(String name);
+
+    /**
+     * Restore cache group(s) from the snapshot.
+     * <p>
+     * <b>NOTE:</b> Cache groups to be restored from the snapshot must not present in the cluster, if they present,
+     * they must be destroyed by the user (eg with {@link IgniteCache#destroy()}) before starting this operation.
+     *
+     * @param name Snapshot name.
+     * @param cacheGroupNames Cache groups to be restored or {@code null} to restore all cache groups from the snapshot.
+     * @return Future which will be completed when restore operation finished.
+     */
+    public IgniteFuture<Void> restoreSnapshot(String name, @Nullable Collection<String> cacheGroupNames);
 }

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

@@ -139,7 +139,10 @@ public enum IgniteFeatures {
     CACHE_GROUP_KEY_CHANGE(47),
 
     /** Collecting performance statistics. */
-    PERFORMANCE_STATISTICS(48);
+    PERFORMANCE_STATISTICS(48),
+
+    /** Restore cache group from the snapshot. */
+    SNAPSHOT_RESTORE_CACHE_GROUP(49);
 
     /**
      * Unique feature identifier.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java

@@ -48,13 +48,15 @@
 import org.apache.ignite.internal.IgniteFeatures;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.IgniteNodeAttributes;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
 import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
 import org.apache.ignite.internal.managers.encryption.GridEncryptionManager;
 import org.apache.ignite.internal.managers.systemview.walker.CacheGroupViewWalker;
 import org.apache.ignite.internal.managers.systemview.walker.CacheViewWalker;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteClusterReadOnlyException;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
 import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
 import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
@@ -582,6 +584,28 @@ public boolean onCacheChangeRequested(DynamicCacheChangeBatch batch, AffinityTop
         DiscoveryDataClusterState state = ctx.state().clusterState();
 
         if (state.active() && !state.transition()) {
+            Set<IgniteUuid> restartIds = new HashSet<>(F.viewReadOnly(
+                batch.requests(), DynamicCacheChangeRequest::restartId, req -> req.start() && req.restartId() != null));
+
+            assert restartIds.size() <= 1 : batch.requests();
+
+            Collection<UUID> nodes = ctx.cache().context().snapshotMgr().cacheStartRequiredAliveNodes(F.first(restartIds));
+
+            for (UUID nodeId : nodes) {
+                ClusterNode node = ctx.discovery().node(nodeId);
+
+                if (node != null && CU.baselineNode(node, state) && ctx.discovery().alive(node))
+                    continue;
+
+                ClusterTopologyCheckedException err =
+                    new ClusterTopologyCheckedException("Required node has left the cluster [nodeId=" + nodeId + ']');
+
+                for (DynamicCacheChangeRequest req : batch.requests())
+                    ctx.cache().completeCacheStartFuture(req, false, err);
+
+                return false;
+            }
+
             ExchangeActions exchangeActions = new ExchangeActions();
 
             CacheChangeProcessResult res = processCacheChangeRequests(exchangeActions,
@@ -593,6 +617,9 @@ public boolean onCacheChangeRequested(DynamicCacheChangeBatch batch, AffinityTop
                 assert !exchangeActions.empty() : exchangeActions;
 
                 batch.exchangeActions(exchangeActions);
+
+                if (!nodes.isEmpty())
+                    exchangeActions.cacheStartRequiredAliveNodes(nodes);
             }
 
             return res.needExchange;
@@ -1007,6 +1034,16 @@ else if (encMgr.masterKeyDigest() != null &&
             }
         }
 
+        if (err == null && req.restartId() == null) {
+            IgniteSnapshotManager snapshotMgr = ctx.cache().context().snapshotMgr();
+
+            if (snapshotMgr.isRestoring(cacheName, ccfg.getGroupName())) {
+                err = new IgniteCheckedException("Cache start failed. A cache or group with the same name is " +
+                    "currently being restored from a snapshot [cache=" + cacheName +
+                    (ccfg.getGroupName() == null ? "" : ", group=" + ccfg.getGroupName()) + ']');
+            }
+        }
+
         if (err != null) {
             if (persistedCfgs)
                 res.errs.add(err);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeActions.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import org.apache.ignite.internal.util.typedef.C1;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -43,6 +44,12 @@ public class ExchangeActions {
     /** */
     private Map<String, CacheActionData> cachesToStart;
 
+    /**
+     * Server nodes on which a successful start of the cache(s) is required, if any of these nodes fails when starting
+     * the cache(s), the whole procedure is rolled back.
+     */
+    private Collection<UUID> cacheStartRequiredAliveNodes;
+
     /** */
     private Map<String, CacheActionData> cachesToStop;
 
@@ -319,6 +326,23 @@ public boolean cacheGroupStarting(int grpId) {
         return false;
     }
 
+    /**
+     * @return Server nodes on which a successful start of the cache(s) is required, if any of these nodes fails when
+     *      starting the cache(s), the whole procedure is rolled back.
+     */
+    public Collection<UUID> cacheStartRequiredAliveNodes() {
+        return cacheStartRequiredAliveNodes == null ? Collections.emptyList() : cacheStartRequiredAliveNodes;
+    }
+
+    /**
+     * @param cacheStartRequiredAliveNodes Server nodes on which a successful start of the cache(s) is required, if any
+     *                                     of these nodes fails when starting the cache(s), the whole procedure is
+     *                                     rolled back.
+     */
+    public void cacheStartRequiredAliveNodes(Collection<UUID> cacheStartRequiredAliveNodes) {
+        this.cacheStartRequiredAliveNodes = new ArrayList<>(cacheStartRequiredAliveNodes);
+    }
+
     /**
      * @param grpDesc Group descriptor.
      * @param destroy Destroy flag.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java

@@ -4268,6 +4268,9 @@ else if (msg0 instanceof WalStateFinishMessage)
         if (res == null)
             res = validateRestartingCaches(node);
 
+        if (res == null)
+            res = validateRestoringCaches(node);
+
         return res;
     }
 
@@ -4294,6 +4297,20 @@ private IgniteNodeValidationResult validateRestartingCaches(ClusterNode node) {
         return null;
     }
 
+    /**
+     * @param node Joining node to validate.
+     * @return Node validation result if there was an issue with the joining node, {@code null} otherwise.
+     */
+    private IgniteNodeValidationResult validateRestoringCaches(ClusterNode node) {
+        if (ctx.cache().context().snapshotMgr().isRestoring()) {
+            String msg = "Joining node during caches restore is not allowed [joiningNodeId=" + node.id() + ']';
+
+            return new IgniteNodeValidationResult(node.id(), msg);
+        }
+
+        return null;
+    }
+
     /**
      * @return Keep static cache configuration flag. If {@code true}, static cache configuration will override
      * configuration persisted on disk.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java

@@ -293,6 +293,8 @@ public GridCacheSharedContext(
 
         stateAwareMgrs.add(snpMgr);
 
+        stateAwareMgrs.add(snapshotMgr);
+
         for (PluginProvider prv : kernalCtx.plugins().allProviders())
             if (prv instanceof IgniteChangeGlobalStateSupport)
                 stateAwareMgrs.add(((IgniteChangeGlobalStateSupport)prv));

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java

@@ -32,6 +32,7 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.function.BooleanSupplier;
 import javax.cache.CacheException;
 import org.apache.ignite.IgniteBinary;
 import org.apache.ignite.IgniteCheckedException;
@@ -55,6 +56,7 @@
 import org.apache.ignite.internal.IgniteFeatures;
 import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
 import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.IgniteNodeAttributes;
 import org.apache.ignite.internal.NodeStoppingException;
 import org.apache.ignite.internal.UnregisteredBinaryTypeException;
@@ -992,6 +994,42 @@ public BinaryMetadata binaryMetadata(int typeId) throws BinaryObjectException {
         }
     }
 
+    /** {@inheritDoc} */
+    @Override public void updateMetadata(File metadataDir, BooleanSupplier stopChecker) throws IgniteCheckedException {
+        if (!metadataDir.exists())
+            return;
+
+        try {
+            ConcurrentMap<Integer, BinaryMetadataHolder> metaCache = new ConcurrentHashMap<>();
+
+            new BinaryMetadataFileStore(metaCache, ctx, log, metadataDir)
+                .restoreMetadata();
+
+            Collection<BinaryMetadata> metadata = F.viewReadOnly(metaCache.values(), BinaryMetadataHolder::metadata);
+
+            // Check the compatibility of the binary metadata.
+            for (BinaryMetadata newMeta : metadata) {
+                BinaryMetadata oldMeta = binaryMetadata(newMeta.typeId());
+
+                if (oldMeta != null)
+                    BinaryUtils.mergeMetadata(oldMeta, newMeta, null);
+            }
+
+            // Update cluster metadata.
+            for (BinaryMetadata newMeta : metadata) {
+                if (stopChecker.getAsBoolean())
+                    return;
+
+                if (Thread.interrupted())
+                    throw new IgniteInterruptedCheckedException("Thread has been interrupted.");
+
+                addMeta(newMeta.typeId(), newMeta.wrap(binaryContext()), false);
+            }
+        } catch (BinaryObjectException e) {
+            throw new IgniteCheckedException(e);
+        }
+    }
+
     /** {@inheritDoc} */
     @Override public BinaryObject buildEnum(String typeName, int ord) throws BinaryObjectException {
         A.notNullOrEmpty(typeName, "enum type name");

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java

@@ -299,7 +299,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     private boolean forceAffReassignment;
 
     /** Exception that was thrown during init phase on local node. */
-    private Exception exchangeLocE;
+    private volatile Exception exchangeLocE;
 
     /** Exchange exceptions from all participating nodes. */
     private final Map<UUID, Exception> exchangeGlobalExceptions = new ConcurrentHashMap<>();
@@ -5126,6 +5126,12 @@ public void onNodeLeft(final ClusterNode node) {
 
                             if (crd0 == null)
                                 finishState = new FinishState(null, initialVersion(), null);
+
+                            if (dynamicCacheStartExchange() && exchangeLocE == null &&
+                                exchActions.cacheStartRequiredAliveNodes().contains(node.id())) {
+                                exchangeGlobalExceptions.put(cctx.localNodeId(), exchangeLocE = new ClusterTopologyCheckedException(
+                                    "Required node has left the cluster [nodeId=" + node.id() + ']'));
+                            }
                         }
 
                         if (crd0 == null) {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java

@@ -974,29 +974,37 @@ public void readConfigurationFiles(List<CacheConfiguration<?, ?>> ccfgs,
         Arrays.sort(files);
 
         for (File file : files) {
-            if (file.isDirectory()) {
-                if (file.getName().startsWith(CACHE_DIR_PREFIX)) {
-                    File conf = new File(file, CACHE_DATA_FILENAME);
+            if (file.isDirectory())
+                readCacheConfigurations(file, ccfgs);
+        }
 
-                    if (conf.exists() && conf.length() > 0) {
-                        StoredCacheData cacheData = readCacheData(conf);
+        return ccfgs;
+    }
 
-                        String cacheName = cacheData.config().getName();
+    /**
+     * @param dir Cache (group) directory.
+     * @param ccfgs Cache configurations.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void readCacheConfigurations(File dir, Map<String, StoredCacheData> ccfgs) throws IgniteCheckedException {
+        if (dir.getName().startsWith(CACHE_DIR_PREFIX)) {
+            File conf = new File(dir, CACHE_DATA_FILENAME);
 
-                        if (!ccfgs.containsKey(cacheName))
-                            ccfgs.put(cacheName, cacheData);
-                        else {
-                            U.warn(log, "Cache with name=" + cacheName + " is already registered, skipping config file "
-                                    + file.getName());
-                        }
-                    }
+            if (conf.exists() && conf.length() > 0) {
+                StoredCacheData cacheData = readCacheData(conf);
+
+                String cacheName = cacheData.config().getName();
+
+                if (!ccfgs.containsKey(cacheName))
+                    ccfgs.put(cacheName, cacheData);
+                else {
+                    U.warn(log, "Cache with name=" + cacheName + " is already registered, skipping config file "
+                        + dir.getName());
                 }
-                else if (file.getName().startsWith(CACHE_GRP_DIR_PREFIX))
-                    readCacheGroupCaches(file, ccfgs);
             }
         }
-
-        return ccfgs;
+        else if (dir.getName().startsWith(CACHE_GRP_DIR_PREFIX))
+            readCacheGroupCaches(dir, ccfgs);
     }
 
     /**