You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by al...@apache.org on 2025/04/25 07:09:11 UTC
(ignite) branch master updated: IGNITE-25194 Fix corrupted tree exception on expiration - Fixes #12020.
This is an automated email from the ASF dual-hosted git repository.
alexpl 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 93d896b3d76 IGNITE-25194 Fix corrupted tree exception on expiration - Fixes #12020.
93d896b3d76 is described below
commit 93d896b3d76d0a687155a95dc9103090768c21f5
Author: Aleksey Plekhanov <pl...@gmail.com>
AuthorDate: Fri Apr 25 10:06:25 2025 +0300
IGNITE-25194 Fix corrupted tree exception on expiration - Fixes #12020.
Signed-off-by: Aleksey Plekhanov <pl...@gmail.com>
---
.../processors/cache/GridCacheMapEntry.java | 19 ++--
.../persistence/PendingTreeCorruptionTest.java | 112 ++++++++++++++++++++-
2 files changed, 123 insertions(+), 8 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index fba18aff20d..da6f47073df 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -4954,13 +4954,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
newExpireTime = entry.expireTimeExtras();
}
else if (newSysTtl == CU.TTL_ZERO) {
- op = DELETE;
-
- writeObj = null;
-
- remove(conflictCtx, invokeRes, readFromStore, false);
-
- return;
+ newTtl = CU.TTL_ZERO;
+ newSysExpireTime = newExpireTime = U.currentTimeMillis() - 1; // Force expiration.
}
else {
newSysExpireTime = CU.EXPIRE_TIME_CALCULATE;
@@ -4974,6 +4969,16 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
newSysExpireTime = newExpireTime = conflictCtx.expireTime();
}
+ if (newExpireTime > 0 && newExpireTime < U.currentTimeMillis()) {
+ op = DELETE;
+
+ writeObj = null;
+
+ remove(conflictCtx, invokeRes, readFromStore, false);
+
+ return;
+ }
+
if (intercept && (conflictVer == null || !skipInterceptorOnConflict)) {
Object updated0 = cctx.unwrapBinaryIfNeeded(updated, keepBinary, false, null);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/PendingTreeCorruptionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/PendingTreeCorruptionTest.java
index 3337fb6f2d3..41fa0988021 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/PendingTreeCorruptionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/PendingTreeCorruptionTest.java
@@ -17,10 +17,15 @@
package org.apache.ignite.internal.processors.cache.persistence;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.cache.expiry.AccessedExpiryPolicy;
import javax.cache.expiry.Duration;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cluster.ClusterState;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
@@ -28,20 +33,35 @@ 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.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
import org.apache.ignite.internal.processors.cache.IgniteCacheOffheapManager;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
+import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
+import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.tree.PendingEntriesTree;
import org.apache.ignite.internal.processors.cache.tree.PendingRow;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.lang.GridCursor;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
/** */
public class PendingTreeCorruptionTest extends GridCommonAbstractTest {
+ /** PDS enabled. */
+ private boolean pds;
+
/** */
@Before
public void before() throws Exception {
@@ -66,7 +86,7 @@ public class PendingTreeCorruptionTest extends GridCommonAbstractTest {
cfg.setDataStorageConfiguration(new DataStorageConfiguration()
.setDefaultDataRegionConfiguration(new DataRegionConfiguration()
- .setPersistenceEnabled(true)
+ .setPersistenceEnabled(pds)
)
.setWalSegments(3)
.setWalSegmentSize(512 * 1024)
@@ -78,6 +98,8 @@ public class PendingTreeCorruptionTest extends GridCommonAbstractTest {
/** */
@Test
public void testCorruptionWhileLoadingData() throws Exception {
+ pds = true;
+
IgniteEx ig = startGrid(0);
ig.cluster().state(ClusterState.ACTIVE);
@@ -157,4 +179,92 @@ public class PendingTreeCorruptionTest extends GridCommonAbstractTest {
ig.context().cache().context().database().checkpointReadUnlock();
}
}
+
+ /** */
+ @Test
+ @WithSystemProperty(key = IgniteSystemProperties.IGNITE_UNWIND_THROTTLING_TIMEOUT, value = "1000")
+ public void testCorruptionOnExpiration() throws Exception {
+ pds = false;
+
+ IgniteEx srv = startGrid();
+
+ CountDownLatch expirationStarted = new CountDownLatch(1);
+ CountDownLatch entryUpdated = new CountDownLatch(1);
+
+ IgniteCache<Object, Object> cache = srv.getOrCreateCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setAffinity(new ExpirationWaitingRendezvousAffinityFunction(expirationStarted, entryUpdated)));
+
+ // Warmup to ensure that the next put/remove/put will create row with the same link.
+ cache.put(0, 0);
+ cache.remove(0);
+
+ IgniteInternalCache<Object, Object> cachex = srv.cachex(DEFAULT_CACHE_NAME);
+
+ GridCacheVersion ver = new GridCacheVersion(1, 1, 1, 2);
+ KeyCacheObject key = new KeyCacheObjectImpl(0, null, -1);
+ CacheObjectImpl val = new CacheObjectImpl(0, null);
+ GridCacheDrInfo drInfo = new GridCacheDrExpirationInfo(val, ver, 1, CU.toExpireTime(1000));
+
+ Map<KeyCacheObject, GridCacheDrInfo> map = F.asMap(key, drInfo);
+
+ cachex.putAllConflict(map);
+
+ // Wait for PendingTree row removal.
+ assertTrue(expirationStarted.await(10, SECONDS));
+
+ // Remove entry and put entry with the same key, with the same expire time to the same place (with the same link).
+ cachex.removeAllConflict(F.asMap(key, ver));
+ cachex.putAllConflict(map);
+
+ // Resume expiration thread.
+ entryUpdated.countDown();
+
+ // Wait for entry removal by expiration.
+ assertTrue(GridTestUtils.waitForCondition(() -> !cache.containsKey(0), 1_000L));
+
+ // Check pending tree is in consistent state.
+ CacheGroupContext grp = cachex.context().group();
+ PendingEntriesTree pendingTree = grp.topology().localPartition(0).dataStore().pendingTree();
+
+ int cacheId = CU.cacheId(DEFAULT_CACHE_NAME);
+
+ List<PendingRow> rows = pendingTree.remove(new PendingRow(cacheId, Long.MIN_VALUE, 0),
+ new PendingRow(cacheId, U.currentTimeMillis(), 0), 1);
+
+ assertTrue(rows.isEmpty());
+ }
+
+ /** */
+ @SuppressWarnings("TransientFieldNotInitialized")
+ private static class ExpirationWaitingRendezvousAffinityFunction extends RendezvousAffinityFunction {
+ /** */
+ private final transient CountDownLatch expirationStarted;
+
+ /** */
+ private final transient CountDownLatch entryUpdated;
+
+ /** */
+ public ExpirationWaitingRendezvousAffinityFunction(
+ CountDownLatch expirationStarted,
+ CountDownLatch entryUpdated
+ ) {
+ this.expirationStarted = expirationStarted;
+ this.entryUpdated = entryUpdated;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int partition(Object key) {
+ // Partition calculation is invoked during expiration of each key for in-memory cluster.
+ // Inject code to simulate race here.
+ if (Thread.currentThread().getName().contains("ttl-cleanup-worker")) {
+ expirationStarted.countDown();
+
+ // Suspend ttl-cleanup-worker after PendingTree row is removed, but before the corresponding
+ // expired row is deleted from cache data tree and row store.
+ U.awaitQuiet(entryUpdated);
+ }
+
+ return super.partition(key);
+ }
+ }
}