IGNITE-28457 Fix flaky WalCompactionSwitchOnTest#testWalCompactionSwitch

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/wal/WalCompactionSwitchOnTest.java

@@ -18,21 +18,25 @@
 package org.apache.ignite.internal.processors.cache.persistence.db.wal;
 
 import java.io.File;
+import java.util.Comparator;
+import java.util.Objects;
+import java.util.stream.Stream;
 import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CacheMode;
 import org.apache.ignite.cluster.ClusterState;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.DataRegionConfiguration;
 import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree;
-import org.apache.ignite.internal.util.lang.GridAbsPredicate;
-import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
 import org.junit.Test;
 
+import static org.apache.ignite.configuration.DataStorageConfiguration.UNLIMITED_WAL_ARCHIVE;
 import static org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager.WAL_SEGMENT_TEMP_FILE_COMPACTED_FILTER;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
 
 /**
  * Load without compaction -> Stop -> Enable WAL Compaction -> Start.
@@ -42,84 +46,107 @@
     private boolean compactionEnabled;
 
     /** {@inheritDoc} */
-    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
-        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
-
-        cfg.setDataStorageConfiguration(new DataStorageConfiguration()
-                .setDefaultDataRegionConfiguration(
-                        new DataRegionConfiguration()
-                            .setPersistenceEnabled(true)
-                            .setMaxSize(256 * 1024 * 1024))
-                .setWalSegmentSize(512 * 1024)
-                .setWalSegments(100)
-                .setWalCompactionEnabled(compactionEnabled));
-
-        return cfg;
+    @Override protected void beforeTest() throws Exception {
+        super.beforeTest();
+
+        stopAllGrids();
+
+        cleanPersistenceDir();
     }
 
     /** {@inheritDoc} */
-    @Override protected void beforeTest() throws Exception {
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        stopAllGrids();
+
         cleanPersistenceDir();
     }
 
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        DataStorageConfiguration dsCfg = new DataStorageConfiguration()
+            .setDefaultDataRegionConfiguration(
+                new DataRegionConfiguration()
+                    .setPersistenceEnabled(true)
+                    .setMaxSize(256 * 1024 * 1024)
+            )
+            .setWalSegmentSize(512 * 1024)
+            .setWalSegments(100)
+            .setMaxWalArchiveSize(UNLIMITED_WAL_ARCHIVE)
+            .setWalCompactionEnabled(compactionEnabled);
+
+        return super.getConfiguration(igniteInstanceName)
+            .setDataStorageConfiguration(dsCfg)
+            .setCacheConfiguration(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+    }
+
     /**
      * Load without compaction -> Stop -> Enable WAL Compaction -> Start.
      *
      * @throws Exception On exception.
      */
     @Test
     public void testWalCompactionSwitch() throws Exception {
-        IgniteEx ex = startGrid(0);
-
-        ex.cluster().state(ClusterState.ACTIVE);
+        IgniteEx n = startGrid(0);
+        n.cluster().state(ClusterState.ACTIVE);
 
-        IgniteCache<Integer, Integer> cache = ex.getOrCreateCache(
-            new CacheConfiguration<Integer, Integer>()
-                    .setName("c1")
-                    .setGroupName("g1")
-                    .setCacheMode(CacheMode.PARTITIONED)
-        );
+        IgniteCache<Integer, Integer> cache = n.cache(DEFAULT_CACHE_NAME);
+        for (int i = 0; i < 10_000; i++)
+            cache.put(i, i); // Streamer is intentionally not used to ensure there are more WalRecords.
 
-        for (int i = 0; i < 500; i++)
-            cache.put(i, i);
-
-        NodeFileTree ft = ex.context().pdsFolderResolver().fileTree();
+        File walArchiveDir = fileTree(n).walArchive();
+        assertNotNull(walArchiveDir);
 
         forceCheckpoint();
-
-        GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                File[] archivedFiles = ft.walSegments();
-
-                return archivedFiles.length == 39;
-            }
-        }, 5000);
+        assertTrue(waitForCondition(() -> collectAndSortWalSegments(walArchiveDir).length >= 10, 5_000, 32));
 
         stopGrid(0);
 
+        assertTrue(collectAndSortWalSegments(walArchiveDir).length >= 10);
+        assertTrue(collectAndSortCompactedWalSegments(walArchiveDir).length == 0);
+
         compactionEnabled = true;
+        startGrid(0);
 
-        ex = startGrid(0);
+        assertTrue(waitForCondition(() -> collectAndSortCompactedWalSegments(walArchiveDir).length >= 5, 5_000, 32));
 
-        ex.cluster().state(ClusterState.ACTIVE);
+        File minTmpFile = minTmpWalSegemnt(walArchiveDir);
+        // Checks that tmp files of compacted WAL segments are missing or have been deleted for compacted ones.
+        assertTrue(Objects.toString(minTmpFile), minTmpFile == null || walSegmentIdx(minTmpFile) >= 5);
+    }
 
-        File archiveDir = ex.context().pdsFolderResolver().fileTree().walArchive();
+    /** */
+    private static NodeFileTree fileTree(IgniteEx n) {
+        return n.context().pdsFolderResolver().fileTree();
+    }
 
-        GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                File[] archivedFiles = archiveDir.listFiles(NodeFileTree::walCompactedSegment);
+    /** */
+    private static @Nullable File minTmpWalSegemnt(File dir) {
+        return Stream.of(dir.listFiles(WAL_SEGMENT_TEMP_FILE_COMPACTED_FILTER))
+            .min(Comparator.comparingLong(WalCompactionSwitchOnTest::walSegmentIdx))
+            .orElse(null);
+    }
 
-                return archivedFiles.length == 20;
-            }
-        }, 5000);
+    /** */
+    private static File[] collectAndSortWalSegments(File dir) {
+        return sortWalSegments(dir.listFiles(f -> NodeFileTree.walSegment(f)));
+    }
 
-        File[] tmpFiles = archiveDir.listFiles(WAL_SEGMENT_TEMP_FILE_COMPACTED_FILTER);
+    /** */
+    private static File[] collectAndSortCompactedWalSegments(File dir) {
+        return sortWalSegments(dir.listFiles(NodeFileTree::walCompactedSegment));
+    }
 
-        assertEquals(0, tmpFiles.length);
+    /** */
+    private static File[] sortWalSegments(File... walSegments) {
+        return Stream.of(walSegments)
+            .sorted(Comparator.comparingLong(f -> U.fixedLengthFileNumber(f.getName())))
+            .toArray(File[]::new);
     }
 
-    /** {@inheritDoc} */
-    @Override protected void afterTest() throws Exception {
-        stopAllGrids();
+    /** */
+    private static long walSegmentIdx(File f) {
+        return U.fixedLengthFileNumber(f.getName());
     }
 }