IGNITE-9420 Move logical recovery outside of PME - Fixes #5067.

Signed-off-by: Alexey Goncharuk <alexey.goncharuk@gmail.com>

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

@@ -698,4 +698,9 @@ public interface GridKernalContext extends Iterable<GridComponent> {
      * @return Default uncaught exception handler used by thread pools.
      */
     public Thread.UncaughtExceptionHandler uncaughtExceptionHandler();
+
+    /**
+     * @return {@code True} if node is in recovery mode (before join to topology).
+     */
+    public boolean recoveryMode();
 }

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

@@ -415,6 +415,9 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
     /** Failure processor. */
     private FailureProcessor failureProc;
 
+    /** Recovery mode flag. Flag is set to {@code false} when discovery manager started. */
+    private boolean recoveryMode = true;
+
     /**
      * No-arg constructor is required by externalization.
      */
@@ -1180,6 +1183,18 @@ void disconnected(boolean disconnected) {
         return hnd;
     }
 
+    /** {@inheritDoc} */
+    @Override public boolean recoveryMode() {
+        return recoveryMode;
+    }
+
+    /**
+     * @param recoveryMode Recovery mode.
+     */
+    public void recoveryMode(boolean recoveryMode) {
+        this.recoveryMode = recoveryMode;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridKernalContextImpl.class, this);

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

@@ -1048,6 +1048,8 @@ public void start(
                 ctx.cache().context().database().notifyMetaStorageSubscribersOnReadyForRead();
 
                 ctx.cache().context().database().startMemoryRestore(ctx);
+
+                ctx.recoveryMode(false);
             }
             catch (Throwable e) {
                 U.error(

modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java

@@ -316,6 +316,9 @@ public void record(DiscoveryEvent evt, DiscoCache discoCache) {
     private void record0(Event evt, Object... params) {
         assert evt != null;
 
+        if (ctx.recoveryMode())
+            return;
+
         if (!enterBusy())
             return;
 

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/DataRecord.java

@@ -76,6 +76,13 @@ public DataRecord(List<DataEntry> writeEntries, long timestamp) {
         this.writeEntries = writeEntries;
     }
 
+    /**
+     * @param writeEntries Write entries.
+     */
+    public void setWriteEntries(List<DataEntry> writeEntries) {
+        this.writeEntries = writeEntries;
+    }
+
     /**
      * @return Collection of write entries.
      */

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/MemoryRecoveryRecord.java

@@ -20,16 +20,11 @@
 import org.apache.ignite.internal.util.typedef.internal.S;
 
 /**
- * Marker that we start memory recovering.
- *
- * @deprecated Previously, used to track node started\stopped states. But in fact only
- * mark files created by method GridCacheDatabaseSharedManager#nodeStart(WALPointer)
- * used. Should be removed in 3.0 release.
+ * Marker indicates that binary memory recovery has finished.
  */
-@Deprecated
 public class MemoryRecoveryRecord extends WALRecord {
     /** Create timestamp, millis */
-    private long time;
+    private final long time;
 
     /**
      * Default constructor.

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/MetastoreDataRecord.java

@@ -18,13 +18,14 @@
 
 package org.apache.ignite.internal.pagemem.wal.record;
 
+import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.jetbrains.annotations.Nullable;
 
 /**
  *
  */
-public class MetastoreDataRecord extends WALRecord {
+public class MetastoreDataRecord extends WALRecord implements WalRecordCacheGroupAware {
     /** */
     private final String key;
 
@@ -59,4 +60,9 @@ public String key() {
     @Override public String toString() {
         return S.toString(MetastoreDataRecord.class, this, "super", super.toString());
     }
+
+    /** {@inheritDoc} */
+    @Override public int groupId() {
+        return MetaStorage.METASTORAGE_CACHE_ID;
+    }
 }

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/MetaPageUpdatePartitionDataRecord.java

@@ -55,7 +55,8 @@ public MetaPageUpdatePartitionDataRecord(
         long updateCntr,
         long globalRmvId,
         int partSize,
-        long cntrsPageId, byte state,
+        long cntrsPageId,
+        byte state,
         int allocatedIdxCandidate
     ) {
         super(grpId, pageId);

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

@@ -17,7 +17,6 @@
 
 package org.apache.ignite.internal.processors.cache;
 
-import javax.cache.CacheException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -32,6 +31,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Collectors;
+import javax.cache.CacheException;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteSystemProperties;
@@ -362,11 +362,7 @@ public Set<Integer> waitGroups() {
     void onCacheGroupCreated(CacheGroupContext grp) {
         if (!grpHolders.containsKey(grp.groupId())) {
             cctx.io().addCacheGroupHandler(grp.groupId(), GridDhtAffinityAssignmentResponse.class,
-                new IgniteBiInClosure<UUID, GridDhtAffinityAssignmentResponse>() {
-                    @Override public void apply(UUID nodeId, GridDhtAffinityAssignmentResponse res) {
-                        processAffinityAssignmentResponse(nodeId, res);
-                    }
-                });
+                (IgniteBiInClosure<UUID, GridDhtAffinityAssignmentResponse>) this::processAffinityAssignmentResponse);
         }
     }
 
@@ -1281,6 +1277,7 @@ private void forAllCacheGroups(boolean crd, IgniteInClosureX<GridAffinityAssignm
         else {
             affinityCaches = cctx.kernalContext().cache().cacheGroups().stream()
                 .filter(grp -> !grp.isLocal())
+                .filter(grp -> !grp.isRecoveryMode())
                 .map(CacheGroupContext::affinity)
                 .collect(Collectors.toList());
         }