IGNITE-14070 Snapshot listeners management.

Author: xtern

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

@@ -101,6 +101,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadWriteMetastorage;
 import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.lifecycle.SnapshotLifecycleListener;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.lifecycle.SnapshotListeners;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPagePayload;
 import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
@@ -303,7 +304,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
     private ClusterSnapshotFuture clusterSnpFut;
 
     // todo something like pipeline
-    private final List<SnapshotLifecycleListener> lifecycleListeners = new ArrayList<>();
+    private SnapshotListeners snpLsnrs;
 
     /** Current snapshot operation on local node. */
     private volatile SnapshotOperationRequest clusterSnpReq;
@@ -385,12 +386,10 @@ public static String partDeltaFileName(int partId) {
         U.ensureDirectory(locSnpDir, "snapshot work directory", log);
         U.ensureDirectory(tmpWorkDir, "temp directory for snapshot creation", log);
 
-        lifecycleListeners.add(new SnapshotLifeCycleListenerImpl(ctx));
+        snpLsnrs = new SnapshotListeners(ctx.plugins());
 
-        SnapshotLifecycleListener[] lsnrs = cctx.kernalContext().plugins().extensions(SnapshotLifecycleListener.class);
-
-        if (lsnrs != null)
-            Collections.addAll(lifecycleListeners, lsnrs);
+        // Register default.
+        snpLsnrs.register(new SnapshotRestoreConsistencyCheck(ctx));
 
         MetricRegistry mreg = cctx.kernalContext().metric().registry(SNAPSHOT_METRICS);
 
@@ -549,8 +548,8 @@ public void deleteSnapshot(File snpDir, String folderName) {
         }
     }
 
-    public Collection<SnapshotLifecycleListener> lifeCycleListeners() {
-        return lifecycleListeners;
+    public SnapshotListeners listeners() {
+        return snpLsnrs;
     }
 
     /**
@@ -691,7 +690,7 @@ private IgniteInternalFuture<SnapshotOperationResponse> initLocalSnapshotStartSt
                     log.info("Snapshot metafile has been created: " + smf.getAbsolutePath());
                 }
 
-                for (SnapshotLifecycleListener lsnr : lifecycleListeners)
+                for (SnapshotLifecycleListener lsnr : listeners().list())
                     lsnr.postCreate(req.snapshotName());
 
                 return new SnapshotOperationResponse();

…pshot/SnapshotLifeCycleListenerImpl.java …hot/SnapshotRestoreConsistencyCheck.javamodules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotLifeCycleListenerImpl.java renamed to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreConsistencyCheck.java modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotLifeCycleListenerImpl.java renamed to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreConsistencyCheck.java

@@ -45,14 +45,18 @@
 import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.calculatePartitionHash;
 import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.checkPartitionsPageCrcSum;
 
-public class SnapshotLifeCycleListenerImpl implements SnapshotLifecycleListener {
+public class SnapshotRestoreConsistencyCheck implements SnapshotLifecycleListener {
     /** */
     private final GridKernalContext ctx;
 
     /** */
     private final IgniteLogger log;
 
-    public SnapshotLifeCycleListenerImpl(GridKernalContext ctx) {
+    @Override public int priority() {
+        return Integer.MIN_VALUE;
+    }
+
+    public SnapshotRestoreConsistencyCheck(GridKernalContext ctx) {
         this.ctx = ctx;
 
         log = ctx.log(getClass());

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

@@ -55,7 +55,6 @@
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.ClusterSnapshotFuture;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.lifecycle.RestoreHandleTask;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.lifecycle.SnapshotLifecycleListener;
-import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
 import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
 import org.apache.ignite.internal.util.distributed.DistributedProcess;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -156,7 +155,7 @@ protected void cleanup() throws IgniteCheckedException {
      * @return Future that will be completed when the restore operation is complete and the cache groups are started.
      */
     public IgniteFuture<Void> start(String snpName, @Nullable Collection<String> cacheGrpNames) {
-        ClusterSnapshotFuture fut0;
+        ClusterSnapshotFuture fut0 = null;
 
         try {
             if (ctx.clientNode())
@@ -189,7 +188,9 @@ public IgniteFuture<Void> start(String snpName, @Nullable Collection<String> cac
             Map<ClusterNode, List<SnapshotMetadata>> metas =
                 ctx.cache().context().snapshotMgr().collectSnapshotMetadata(snpName).get();
 
-            Collection<SnapshotLifecycleListener> lsnrs = ctx.cache().context().snapshotMgr().lifeCycleListeners();
+            Collection<SnapshotLifecycleListener> lsnrs = ctx.cache().context().snapshotMgr().listeners().list();
+
+            assert !lsnrs.isEmpty();
 
             if (!lsnrs.isEmpty()) {
                 ctx.security().authorize(ADMIN_SNAPSHOT);
@@ -202,7 +203,6 @@ public IgniteFuture<Void> start(String snpName, @Nullable Collection<String> cac
 
                 ctx.task().execute(RestoreHandleTask.class, new SnapshotPartitionsVerifyTaskArg(cacheGrpNames, metas)).get();
             }
-            //
 
 //            Map<ClusterNode, List<SnapshotMetadata>> metas = f.result().metas();
             Set<UUID> dataNodes = new HashSet<>();
@@ -258,6 +258,9 @@ public IgniteFuture<Void> start(String snpName, @Nullable Collection<String> cac
             prepareRestoreProc.start(req.requestId(), req);
         }
         catch (IgniteException | IgniteCheckedException e) {
+            if (fut0 != null)
+                finishProcess(fut0.rqId, e);
+
             return new IgniteFinishedFutureImpl<>(e);
         }
 

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

@@ -1,3 +1,20 @@
+/*
+ * 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.cache.persistence.snapshot.lifecycle;
 
 import java.util.ArrayList;
@@ -109,7 +126,7 @@ public class RestoreHandleTask extends ComputeTaskAdapter<SnapshotPartitionsVeri
             }
         }
 
-        for (SnapshotLifecycleListener lsnr : ignite.context().cache().context().snapshotMgr().lifeCycleListeners()) {
+        for (SnapshotLifecycleListener lsnr : ignite.context().cache().context().snapshotMgr().listeners().list()) {
             List<ComputeJobResult> nodeResults = resMap.get(lsnr.name());
 
             try {
@@ -149,12 +166,12 @@ public RestoreHandleJob(String snapshotName, String consistentId, Collection<Str
         /** {@inheritDoc} */
         @Override public Map<String, Object> execute() throws IgniteException {
             // todo expand restore hadnler separately
-            Collection<SnapshotLifecycleListener> lsnrs = ignite.context().cache().context().snapshotMgr().lifeCycleListeners();
-
             Map<String, Object> resMap = new HashMap<>();
 
+            SnapshotListeners lsnrs = ignite.context().cache().context().snapshotMgr().listeners();
+
             try {
-                for (SnapshotLifecycleListener lsnr : lsnrs) {
+                for (SnapshotLifecycleListener lsnr : lsnrs.list()) {
                     Object res = lsnr.handleRestore(snapshotName, consistentId, groupNames);
 
                     resMap.put(lsnr.name(), res);

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

@@ -1,3 +1,20 @@
+/*
+ * 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.cache.persistence.snapshot.lifecycle;
 
 import java.util.Collection;
@@ -7,8 +24,14 @@
 import org.apache.ignite.plugin.Extension;
 
 public interface SnapshotLifecycleListener extends Extension {
+    public static final int DEFAULT_PRIORITY = 0;
+
     public String name();
 
+    public default int priority() {
+        return DEFAULT_PRIORITY;
+    }
+
     public default void postCreate(String snpName) throws IgniteCheckedException {
         // No-op.
     }

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

@@ -0,0 +1,90 @@
+/*
+ * 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.cache.persistence.snapshot.lifecycle;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.internal.processors.plugin.IgnitePluginProcessor;
+import org.jetbrains.annotations.Nullable;
+
+public class SnapshotListeners {
+    private final IgnitePluginProcessor plugins;
+    private final Map<String, SnapshotLifecycleListener> lsnrsByName = new HashMap<>();
+    private final List<SnapshotLifecycleListener> lsnrsByPriority = new ArrayList<>();
+
+    public SnapshotListeners(IgnitePluginProcessor plugins) {
+        this.plugins = plugins;
+
+        SnapshotLifecycleListener[] lsnrs = plugins.extensions(SnapshotLifecycleListener.class);
+
+        if (lsnrs == null)
+            return;
+
+        for (SnapshotLifecycleListener lsnr : lsnrs)
+            register(lsnr);
+    }
+
+    public Collection<SnapshotLifecycleListener> list() {
+        return lsnrsByPriority;
+    }
+
+    public @Nullable SnapshotLifecycleListener find(String name) {
+        return lsnrsByName.get(name);
+    }
+
+    public void register(SnapshotLifecycleListener lsnr) {
+        if (lsnrsByName.putIfAbsent(lsnr.name(), lsnr) != null)
+            throw new IllegalArgumentException("Listener named " + lsnr.name() + " is already registered.");
+
+        int idx = Collections.binarySearch(lsnrsByPriority, lsnr, Comparator.comparingInt(SnapshotLifecycleListener::priority));
+
+        if (idx < 0)
+            lsnrsByPriority.add(-idx - 1, lsnr);
+        else
+            lsnrsByPriority.add(idx, lsnr);
+    }
+
+    public boolean disable(String name) {
+        if (lsnrsByPriority.removeIf(v -> name.equals(v.name()))) {
+            lsnrsByName.remove(name);
+
+            return true;
+        }
+
+        return false;
+    }
+
+    public boolean enable(String name) {
+        SnapshotLifecycleListener[] lsnrs = plugins.extensions(SnapshotLifecycleListener.class);
+
+        for (SnapshotLifecycleListener lsnr : lsnrs) {
+            if (name.equals(lsnr.name())) {
+                register(lsnr);
+
+                return true;
+            }
+        }
+
+        return false;
+    }
+}

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotRestoreBaseTest.java

@@ -36,7 +36,7 @@ public abstract class IgniteClusterSnapshotRestoreBaseTest extends AbstractSnaps
 
     /** {@inheritDoc} */
     @Override protected <K, V> CacheConfiguration<K, V> txCacheConfig(CacheConfiguration<K, V> ccfg) {
-        return super.txCacheConfig(ccfg).setAffinity(new RendezvousAffinityFunction(false, 8));
+        return super.txCacheConfig(ccfg).setAffinity(new RendezvousAffinityFunction(false, PARTS_NUMBER));
     }
 
     /**