You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ti...@apache.org on 2023/12/22 17:45:53 UTC
(ignite) branch master updated: IGNITE-21135 Fix CdcManager collects MemoryRecoveryRecord before restore state (#11120)
This is an automated email from the ASF dual-hosted git repository.
timoninmaxim pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 23fc12f8858 IGNITE-21135 Fix CdcManager collects MemoryRecoveryRecord before restore state (#11120)
23fc12f8858 is described below
commit 23fc12f88583d78764b350341e4a8e08e7c905fe
Author: Maksim Timonin <ti...@apache.org>
AuthorDate: Fri Dec 22 20:45:46 2023 +0300
IGNITE-21135 Fix CdcManager collects MemoryRecoveryRecord before restore state (#11120)
---
.../org/apache/ignite/internal/cdc/CdcManager.java | 10 +---
.../GridCacheDatabaseSharedManager.java | 3 +
.../java/org/apache/ignite/cdc/CdcManagerTest.java | 69 ++++++++++++++++++++++
3 files changed, 74 insertions(+), 8 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcManager.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcManager.java
index af9bed790e5..262ced12d42 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcManager.java
@@ -25,13 +25,11 @@ import org.apache.ignite.cdc.CdcConsumer;
import org.apache.ignite.internal.pagemem.wal.record.CdcManagerRecord;
import org.apache.ignite.internal.pagemem.wal.record.CdcManagerStopRecord;
import org.apache.ignite.internal.processors.cache.GridCacheSharedManager;
-import org.apache.ignite.internal.processors.cache.persistence.DatabaseLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.processors.cache.persistence.wal.io.FileInput;
import org.apache.ignite.internal.processors.cache.persistence.wal.serializer.RecordSerializer;
-import org.apache.ignite.internal.processors.subscription.GridInternalSubscriptionProcessor;
import org.apache.ignite.lang.IgniteExperimental;
import org.apache.ignite.plugin.PluginContext;
import org.apache.ignite.plugin.PluginProvider;
@@ -65,7 +63,7 @@ import org.apache.ignite.plugin.PluginProvider;
* @see CdcConsumerState#saveCdcMode(CdcMode)
*/
@IgniteExperimental
-public interface CdcManager extends GridCacheSharedManager, DatabaseLifecycleListener {
+public interface CdcManager extends GridCacheSharedManager {
/**
* If this manager isn't enabled then Ignite skips notifying the manager with following methods.
*
@@ -78,16 +76,12 @@ public interface CdcManager extends GridCacheSharedManager, DatabaseLifecycleLis
* It's executed before the first call of {@link #collect(ByteBuffer)}.
* <p> Implementation suggestions:
* <ul>
- * <li>
- * Implementation must subscribe to the database events to get notified with this callback via
- * {@link GridInternalSubscriptionProcessor#registerDatabaseListener(DatabaseLifecycleListener)}
- * </li>
* <li>Callback 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 the Ignite system thread and might get longer the Ignite start procedure.</li>
* <li>Ignite node will fail in case the method throws an exception.</li>
* </ul>
*/
- @Override public default void afterBinaryMemoryRestore(IgniteCacheDatabaseSharedManager mgr,
+ public default void afterBinaryMemoryRestore(IgniteCacheDatabaseSharedManager mgr,
GridCacheDatabaseSharedManager.RestoreBinaryState restoreState) throws IgniteCheckedException {
// No-op.
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index b6e35b706e7..298dfd6233d 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -1102,6 +1102,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
// Wal logging is now available.
cctx.wal().resumeLogging(restored);
+ if (cctx.cdc() != null && cctx.cdc().enabled())
+ cctx.cdc().afterBinaryMemoryRestore(this, binaryState);
+
// Log MemoryRecoveryRecord to make sure that old physical records are not replayed during
// next physical recovery.
checkpointManager.memoryRecoveryRecordPtr(cctx.wal().log(new MemoryRecoveryRecord(U.currentTimeMillis())));
diff --git a/modules/core/src/test/java/org/apache/ignite/cdc/CdcManagerTest.java b/modules/core/src/test/java/org/apache/ignite/cdc/CdcManagerTest.java
index 223fec45c98..cb3a3735b83 100644
--- a/modules/core/src/test/java/org/apache/ignite/cdc/CdcManagerTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cdc/CdcManagerTest.java
@@ -39,6 +39,8 @@ import org.apache.ignite.internal.pagemem.wal.WALIterator;
import org.apache.ignite.internal.pagemem.wal.record.CheckpointRecord;
import org.apache.ignite.internal.pagemem.wal.record.RolloverType;
import org.apache.ignite.internal.processors.cache.GridCacheSharedManagerAdapter;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager;
import org.apache.ignite.internal.processors.cache.persistence.wal.reader.IgniteWalIteratorFactory;
import org.apache.ignite.internal.processors.configuration.distributed.DistributedChangeableProperty;
@@ -284,6 +286,31 @@ public class CdcManagerTest extends GridCommonAbstractTest {
assertEquals(0, ign.cache(DEFAULT_CACHE_NAME).get(0));
}
+ /** */
+ @Test
+ public void testCollectInvokedAfterRestore() throws Exception {
+ IgniteConfiguration cfg = getConfiguration(getTestIgniteInstanceName());
+
+ AtomicBoolean restored = new AtomicBoolean();
+
+ cfg.setPluginProviders(new AbstractTestPluginProvider() {
+ @Override public String name() {
+ return "CdcManagerPluginProvider";
+ }
+
+ @Override public <T> @Nullable T createComponent(PluginContext ctx, Class<T> cls) {
+ if (CdcManager.class.equals(cls))
+ return (T)new DelayedCdcManager(restored);
+
+ return null;
+ }
+ });
+
+ startGrid(cfg);
+
+ assertTrue(restored.get());
+ }
+
/** */
public void checkCdcContentWithRollover(Runnable rollSegment) throws Exception {
for (int i = 0; i < 10_000; i++)
@@ -412,4 +439,46 @@ public class CdcManagerTest extends GridCommonAbstractTest {
return (TestCdcManager)ign.context().cache().context().cdc();
}
}
+
+ /** Test {@link CdcManager} order of calling methods. */
+ protected static class DelayedCdcManager extends GridCacheSharedManagerAdapter implements CdcManager {
+ /** Set to {@code true} after first {@link #collect(ByteBuffer)} call. */
+ private volatile boolean collected;
+
+ /** Set to {@code true} after binary memory restored. */
+ private final AtomicBoolean restored;
+
+ /** */
+ DelayedCdcManager(AtomicBoolean restored) {
+ this.restored = restored;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void collect(ByteBuffer dataBuf) {
+ collected = true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void afterBinaryMemoryRestore(
+ IgniteCacheDatabaseSharedManager mgr,
+ GridCacheDatabaseSharedManager.RestoreBinaryState restoreState
+ ) {
+ try {
+ // Wait if any WALRecord is being written in background and collected.
+ Thread.sleep(5_000);
+
+ assert !collected;
+
+ restored.set(true);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean enabled() {
+ return true;
+ }
+ }
}