IGNITE-13847 GridEncryptionManager#onWalSegmentRemoved should be invoked async

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

@@ -322,6 +322,15 @@ void reserveWalKey(int grpId, int keyId, long walIdx) {
         return null;
     }
 
+    /**
+     * @return {@code True} if any key reserved for WAL reading can be removed.
+     */
+    boolean isReleaseWalKeysRequired(long walIdx) {
+        Iterator<TrackedWalSegment> iter = trackedWalSegments.iterator();
+
+        return iter.hasNext() && iter.next().idx <= walIdx;
+    }
+
     /**
      * Remove all of the segments that are not greater than the specified index.
      *

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

@@ -924,6 +924,16 @@ public void onDestroyPartitionStore(CacheGroupContext grp, int partId) {
      * @param segmentIdx WAL segment index.
      */
     public void onWalSegmentRemoved(long segmentIdx) {
+        if (grpKeys.isReleaseWalKeysRequired(segmentIdx))
+            ctx.getSystemExecutorService().submit(() -> releaseWalKeys(segmentIdx));
+    }
+
+    /**
+     * Cleanup keys reserved for WAL reading.
+     *
+     * @param segmentIdx WAL segment index.
+     */
+    private void releaseWalKeys(long segmentIdx) {
         withMasterKeyChangeReadLock(() -> {
             synchronized (metaStorageMux) {
                 Map<Integer, Set<Integer>> rmvKeys = grpKeys.releaseWalKeys(segmentIdx);

modules/core/src/test/java/org/apache/ignite/internal/encryption/AbstractEncryptionTest.java

@@ -69,6 +69,7 @@
 import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
 import static org.apache.ignite.spi.encryption.keystore.KeystoreEncryptionSpi.CIPHER_ALGO;
 import static org.apache.ignite.spi.encryption.keystore.KeystoreEncryptionSpi.DEFAULT_MASTER_KEY_NAME;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
 
 /**
  * Abstract encryption test.
@@ -344,6 +345,20 @@ protected void loadData(String cacheName, int cnt) {
         info("Load data finished");
     }
 
+    /**
+     * @param node Ignite node.
+     * @param grpId Cache group ID.
+     * @param keysCnt Expected keys count.
+     */
+    protected void checkKeysCount(IgniteEx node, int grpId, int keysCnt, long timeout)
+        throws IgniteInterruptedCheckedException {
+        GridEncryptionManager encMgr = node.context().encryption();
+
+        waitForCondition(() -> encMgr.groupKeyIds(grpId).size() == keysCnt, timeout);
+
+        assertEquals(keysCnt, encMgr.groupKeyIds(grpId).size());
+    }
+
     /**
      * Ensures that all pages of page store have expected encryption key identifier.
      *

modules/core/src/test/java/org/apache/ignite/internal/encryption/CacheGroupKeyChangeTest.java

@@ -531,7 +531,7 @@ public void testNodeWithOlderKeyBecameCoordinator() throws Exception {
         }
 
         // Make sure the previous key has been removed.
-        assertEquals(1, encrMgr0.groupKeyIds(grpId).size());
+        checkKeysCount(node0, grpId, 1, MAX_AWAIT_MILLIS);
         assertEquals(encrMgr1.groupKeyIds(grpId), encrMgr0.groupKeyIds(grpId));
     }
 
@@ -930,8 +930,8 @@ public void testNodeJoinAfterRotation() throws Exception {
             encrMgr1.onWalSegmentRemoved(maxWalIdx);
         }
 
-        assertEquals(1, encrMgr1.groupKeyIds(grpId).size());
-        assertEquals(encrMgr0.groupKeyIds(grpId), encrMgr1.groupKeyIds(grpId));
+        checkKeysCount(grid(GRID_1), grpId, 1, MAX_AWAIT_MILLIS);
+        checkKeysCount(grid(GRID_0), grpId, 1, MAX_AWAIT_MILLIS);
 
         startGrid(GRID_2);
 

modules/core/src/test/java/org/apache/ignite/internal/encryption/CacheGroupReencryptionTest.java

@@ -518,12 +518,12 @@ public void testNotBltNodeJoin() throws Exception {
         for (long segment = startIdx1; segment <= endIdx1; segment++)
             grid(GRID_0).context().encryption().onWalSegmentRemoved(segment);
 
-        assertEquals(1, grid(GRID_0).context().encryption().groupKeyIds(grpId).size());
+        checkKeysCount(grid(GRID_0), grpId, 1, MAX_AWAIT_MILLIS);
 
         for (long segment = startIdx2; segment <= endIdx2; segment++)
             grid(GRID_1).context().encryption().onWalSegmentRemoved(segment);
 
-        assertEquals(1, grid(GRID_1).context().encryption().groupKeyIds(grpId).size());
+        checkKeysCount(grid(GRID_1), grpId, 1, MAX_AWAIT_MILLIS);
     }
 
     /**