Refactor tests

Author: timoninmaxim

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcConsumerState.java

@@ -72,8 +72,7 @@ public class CdcConsumerState {
     public static final String CACHES_STATE_FILE_NAME = "cdc-caches-state" + FILE_SUFFIX;
 
     /**
-     * The file stores state of CDC mode. {@link CdcManager} creates the file and writes into it, {@link CdcMain} only reads it.
-     * Content of the file is a {@link CdcMode} value:
+     * The file stores state of CDC mode. Content of the file is a {@link CdcMode} value:
      * <ul>
      *     <li>{@link CdcMode#CDC_UTILITY_ACTIVE} means that {@link CdcMain} utility captures data.</li>
      *     <li>{@link CdcMode#IGNITE_NODE_ACTIVE} means that {@link CdcManager} captures data within Ignite node.</li>

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java

@@ -53,7 +53,6 @@
 import org.apache.ignite.internal.pagemem.wal.WALIterator;
 import org.apache.ignite.internal.pagemem.wal.record.CdcManagerRecord;
 import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.cache.GridLocalConfigManager;
 import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
 import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver;
@@ -971,11 +970,6 @@ private void ackAsciiLogo() {
         }
     }
 
-    /** @return Path to CDC state directory. */
-    public static File stateDirFile(GridCacheSharedContext<?, ?> cctx) {
-        return new File(((FileWriteAheadLogManager)cctx.wal(true)).walCdcDirectory(), STATE_DIR);
-    }
-
     /** */
     public static String cdcInstanceName(String igniteInstanceName) {
         return "cdc-" + igniteInstanceName;

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcManager.java

@@ -59,10 +59,14 @@
  */
 public interface CdcManager extends GridCacheSharedManager {
     /**
-     * Callback is invoked once, after Ignite restores memory on start-up. It runs within Ignite start thread before
-     * {@link IgniteWriteAheadLogManager} starts writing WAL records.
+     * Callback is invoked once, after Ignite restores memory on start-up. It invokes before {@link IgniteWriteAheadLogManager}
+     * starts writing WAL records amd before the first call of {@link #collect(ByteBuffer)}.
      *
-     * This method invoked before the first call of {@link #collect(ByteBuffer)}.
+     * Implementation suggestions:
+     * <ul>
+     *     <li>This method can be used for restoring CDC state on Ignite node start, collecting missed events from WAL segments.</li>
+     *     <li>Be aware, this method runs in Ignite main thread and might lengthen the Ignite start procedure.</li>
+     * </ul>
      */
     public void afterMemoryRestore() throws IgniteCheckedException;
 
@@ -97,7 +101,7 @@ public interface CdcManager extends GridCacheSharedManager {
     public void collect(ByteBuffer dataBuf);
 
     /**
-     * If this manager isn't active then Ignite skips preparing data for the {@link #collect(ByteBuffer)} method.
+     * If this manager isn't active then Ignite skips calls of {@link #afterMemoryRestore()} and {@link #collect(ByteBuffer)} methods.
      *
      * @return {@code true} if manager is active, otherwise {@code false}.
      */

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcUtilityActiveCdcManager.java

@@ -22,9 +22,10 @@
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.wal.record.CdcManagerStopRecord;
 import org.apache.ignite.internal.processors.cache.GridCacheSharedManagerAdapter;
+import org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager;
 import org.apache.ignite.internal.util.typedef.internal.U;
 
-import static org.apache.ignite.internal.cdc.CdcMain.stateDirFile;
+import static org.apache.ignite.internal.cdc.CdcMain.STATE_DIR;
 
 /**
  * CDC manager that delegates consuming CDC events to the {@link CdcMain} utility.
@@ -35,7 +36,7 @@ public class CdcUtilityActiveCdcManager extends GridCacheSharedManagerAdapter im
 
     /** {@inheritDoc} */
     @Override protected void start0() throws IgniteCheckedException {
-        File stateDir = stateDirFile(cctx);
+        File stateDir = new File(((FileWriteAheadLogManager)cctx.wal(true)).walCdcDirectory(), STATE_DIR);
 
         if (stateDir.exists()) {
             CdcConsumerState state = new CdcConsumerState(log, stateDir.toPath());

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

@@ -1099,7 +1099,7 @@ private RestoreBinaryState restoreBinaryMemory(
             else if (restored != null)
                 U.log(log, "Binary memory state restored at node startup [restoredPtr=" + restored + ']');
 
-            if (cctx.cdc() != null)
+            if (cctx.cdc() != null && cctx.cdc().active())
                 cctx.cdc().afterMemoryRestore();
 
             // Wal logging is now available.

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

@@ -1067,7 +1067,7 @@ public void startMemoryRestore(GridKernalContext kctx, TimeBag startTimer) throw
             if (ptr != null)
                 ptr = ptr.next();
 
-            if (cctx.cdc() != null)
+            if (cctx.cdc() != null && cctx.cdc().active())
                 cctx.cdc().afterMemoryRestore();
 
             cctx.wal(true).startAutoReleaseSegments();

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/filehandle/FileWriteHandleImpl.java

@@ -541,7 +541,7 @@ private void fsyncReadSegment(SegmentedRingByteBuffer.ReadSegment seg) throws Ig
 
         fsync((MappedByteBuffer)buf.buf, off, len);
 
-        if (cctx.cdc() != null) {
+        if (cctx.cdc() != null && cctx.cdc().active()) {
             try {
                 ByteBuffer cdcBuf = buf.buf.asReadOnlyBuffer();
                 cdcBuf.position(off);

modules/core/src/test/java/org/apache/ignite/cdc/CdcIgniteNodeActiveModeTest.java

@@ -46,6 +46,7 @@
 import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
 import org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneGridKernalContext;
 import org.apache.ignite.internal.processors.metric.MetricRegistry;
+import org.apache.ignite.internal.util.lang.RunnableX;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.plugin.AbstractTestPluginProvider;
 import org.apache.ignite.plugin.PluginContext;
@@ -119,7 +120,7 @@ public class CdcIgniteNodeActiveModeTest extends AbstractCdcTest {
 
     /** */
     @Test
-    public void testCdcMetricsIgniteNodeActiveMode() throws Exception {
+    public void testCdcMetrics() throws Exception {
         awaitCdcModeValue(CdcMode.IGNITE_NODE_ACTIVE);
 
         IgniteCache<Integer, User> cache = ign.cache(DEFAULT_CACHE_NAME);
@@ -157,25 +158,7 @@ public void testSwitchToCdcUtilityActiveMode() throws Exception {
 
     /** */
     @Test
-    public void testCdcModeWritesByIgniteNodeOnly() throws Exception {
-        writeCdcManagerStopCdcRecord();
-
-        rollSegment();
-
-        awaitCdcModeValue(CdcMode.CDC_UTILITY_ACTIVE);
-
-        cdcMainFut.cancel();
-
-        cdcMain = createCdc(new UserCdcConsumer(), ign.configuration());
-
-        cdcMainFut = runAsync(cdcMain);
-
-        awaitCdcModeValue(CdcMode.IGNITE_NODE_ACTIVE);
-    }
-
-    /** */
-    @Test
-    public void testSwitchToPreviousSegment() throws Exception {
+    public void testSkipsNodeCommittedData() throws Exception {
         List<Integer> expUsers = new ArrayList<>();
 
         addData(0, 1, null);
@@ -207,7 +190,7 @@ public void testSwitchToPreviousSegment() throws Exception {
 
     /** */
     @Test
-    public void testCleanOldWals() throws Exception {
+    public void testCleanWalsAfterCdcManagerRecord() throws Exception {
         checkCdcSegmentsExists(0, -1);  // No segments stored in CDC dir.
 
         rollSegment();
@@ -240,20 +223,13 @@ public void testCleanOldWals() throws Exception {
 
     /** */
     @Test
-    public void testIteratorSimple() throws Exception {
-        checkIterator(false);
-    }
-
-    /** */
-    @Test
-    public void testIteratorWithCdcRestart() throws Exception {
-        checkIterator(true);
-    }
+    public void testRestoreModeOnRestart() throws Exception {
+        RunnableX restartUtil = () -> {
+            cdcMainFut.cancel();
+            cdcMain = createCdc(cnsmr, getConfiguration(getTestIgniteInstanceName()));
+            cdcMainFut = runAsync(cdcMain);
+        };
 
-    /**
-     * @param restartCdc Restart cdc.
-     */
-    private void checkIterator(boolean restartCdc) throws Exception {
         List<Integer> expUsers = new ArrayList<>();
 
         addData(0, 1, null);
@@ -262,13 +238,7 @@ private void checkIterator(boolean restartCdc) throws Exception {
 
         addData(2, 3, null);
 
-        if (restartCdc) {
-            cdcMainFut.cancel();
-
-            cdcMain = createCdc(cnsmr, getConfiguration(getTestIgniteInstanceName()));
-
-            cdcMainFut = runAsync(cdcMain);
-        }
+        restartUtil.run();
 
         rollSegment();
 
@@ -279,6 +249,8 @@ private void checkIterator(boolean restartCdc) throws Exception {
 
         rollSegment();
 
+        restartUtil.run();
+
         awaitCdcModeValue(CdcMode.CDC_UTILITY_ACTIVE);
 
         checkConsumerData(expUsers);
@@ -338,7 +310,7 @@ private void awaitCdcModeValue(CdcMode expVal) throws IgniteInterruptedCheckedEx
                 ObjectMetric<String> m = GridTestUtils.<StandaloneGridKernalContext>getFieldValue(cdcMain, "kctx")
                     .metric().registry("cdc").findMetric(CDC_MODE);
 
-                return expVal.name().equals(m.value());
+                return m != null && expVal.name().equals(m.value());
             }
             catch (Exception e) {
                 return false;