Skip to content

Commit 28d7909

Browse files
vldpyatkovvldpyatkov
authored
IGNITE-28612 Stale near-cache value may be returned after rollback to savepoint (#13515)
1 parent ff25813 commit 28d7909

4 files changed

Lines changed: 139 additions & 6 deletions

File tree

‎modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -509,6 +509,8 @@ private void removeLocks(
509509
return;
510510

511511
try {
512+
GridCacheVersion obsoleteVer = forSavepoint ? nextVersion() : null;
513+
512514
int keyCnt = -1;
513515

514516
Map<ClusterNode, GridNearUnlockRequest> map = null;
@@ -558,6 +560,9 @@ private void removeLocks(
558560

559561
// Remove candidate from local node first.
560562
if (entry.removeLock(cand.version())) {
563+
if (forSavepoint)
564+
evictNearEntry(entry, obsoleteVer, cand.topologyVersion());
565+
561566
if (primary.isLocal()) {
562567
dht.removeLocks(
563568
primary.id(),

‎modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxSavepointNearCacheVisibilityTest.java‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.ignite.internal.processors.cache.transactions;
19+
20+
import java.util.concurrent.CountDownLatch;
21+
import java.util.concurrent.TimeUnit;
22+
import org.apache.ignite.Ignite;
23+
import org.apache.ignite.IgniteCache;
24+
import org.apache.ignite.cache.CacheAtomicityMode;
25+
import org.apache.ignite.cache.CacheMode;
26+
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
27+
import org.apache.ignite.configuration.CacheConfiguration;
28+
import org.apache.ignite.configuration.IgniteConfiguration;
29+
import org.apache.ignite.configuration.NearCacheConfiguration;
30+
import org.apache.ignite.internal.IgniteInternalFuture;
31+
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
32+
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtUnlockRequest;
33+
import org.apache.ignite.testframework.GridTestUtils;
34+
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
35+
import org.apache.ignite.transactions.Transaction;
36+
import org.junit.Test;
37+
38+
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
39+
import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
40+
41+
/**
42+
* Test checks near cache entry visibility after a transaction rollback to savepoint.
43+
*/
44+
public class TxSavepointNearCacheVisibilityTest extends GridCommonAbstractTest {
45+
/** {@inheritDoc} */
46+
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
47+
return super.getConfiguration(igniteInstanceName)
48+
.setCacheConfiguration(new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME)
49+
.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
50+
.setNearConfiguration(new NearCacheConfiguration<>())
51+
.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
52+
.setCacheMode(CacheMode.PARTITIONED)
53+
.setBackups(1))
54+
.setCommunicationSpi(new TestRecordingCommunicationSpi());
55+
}
56+
57+
/** {@inheritDoc} */
58+
@Override protected void afterTest() throws Exception {
59+
stopAllGrids();
60+
61+
super.afterTest();
62+
}
63+
64+
/**
65+
* @throws Exception If failed.
66+
*/
67+
@Test
68+
public void testRolledBackEntryVisibleWithoutRemoteUnlock() throws Exception {
69+
Ignite ignite0 = startGridsMultiThreaded(2);
70+
Ignite ignite1 = grid(1);
71+
72+
awaitPartitionMapExchange();
73+
74+
IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME);
75+
IgniteCache<Integer, Integer> cache1 = ignite1.cache(DEFAULT_CACHE_NAME);
76+
77+
int node0Key = primaryKey(cache0);
78+
int node1Key = primaryKey(cache1);
79+
80+
cache0.put(node0Key, -1);
81+
cache0.put(node1Key, -1);
82+
83+
TestRecordingCommunicationSpi commSpi = TestRecordingCommunicationSpi.spi(ignite1);
84+
85+
commSpi.blockMessages((node, msg) ->
86+
msg instanceof GridDhtUnlockRequest && node.id().equals(ignite0.cluster().localNode().id()));
87+
88+
CountDownLatch savepointRolledBackLatch = new CountDownLatch(1);
89+
CountDownLatch finishFirstTxLatch = new CountDownLatch(1);
90+
91+
IgniteInternalFuture<?> fut = GridTestUtils.runAsync(() -> {
92+
try (Transaction tx = ignite0.transactions().txStart(PESSIMISTIC, READ_COMMITTED, 30_000, 2)) {
93+
cache0.put(node0Key, 1);
94+
95+
tx.savepoint("sp");
96+
97+
cache0.put(node1Key, 1);
98+
99+
tx.rollbackToSavepoint("sp");
100+
101+
savepointRolledBackLatch.countDown();
102+
103+
assertTrue(finishFirstTxLatch.await(10, TimeUnit.SECONDS));
104+
105+
cache0.put(node1Key, 2);
106+
107+
tx.commit();
108+
}
109+
});
110+
111+
try {
112+
assertTrue(savepointRolledBackLatch.await(10, TimeUnit.SECONDS));
113+
assertTrue(commSpi.waitForBlocked(1, 10_000));
114+
115+
cache1.put(node1Key, 42);
116+
117+
assertFalse(fut.isDone());
118+
assertEquals(Integer.valueOf(42), cache0.get(node1Key));
119+
}
120+
finally {
121+
commSpi.stopBlock();
122+
123+
finishFirstTxLatch.countDown();
124+
}
125+
126+
fut.get(10_000);
127+
128+
assertEquals(Integer.valueOf(2), cache0.get(node1Key));
129+
}
130+
}

‎modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxSavepointParameterizedTest.java‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -280,16 +280,12 @@ public void testRollbackToSavepointReleasesRemoteDhtLockAcquireAgain() throws Ex
280280
}
281281
});
282282

283+
assertTrue(savepointRolledBackLatch.await(10, TimeUnit.SECONDS));
284+
283285
updateKeyFormPrimary(node1Key);
284286

285287
assertFalse(fut.isDone());
286288

287-
// TODO: IGNITE-28612 Entry visibility violation in transactional replication cache with one backup and near.
288-
if (initKeies && useNearCache && backups == 1 && spKeyOnTxInitiator && !replicated) {
289-
assertTrue(GridTestUtils.waitForCondition(() ->
290-
Integer.valueOf(42).equals(cache0.get(node1Key)), 10_000));
291-
}
292-
293289
assertEquals(Integer.valueOf(42), cache0.get(node1Key));
294290

295291
finishFirstTxLatch.countDown();

‎modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite12.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import org.apache.ignite.internal.processors.cache.transactions.TxRecoveryConcurrentTest;
4949
import org.apache.ignite.internal.processors.cache.transactions.TxRecoveryWithConcurrentRollbackTest;
5050
import org.apache.ignite.internal.processors.cache.transactions.TxSavepointItTest;
51+
import org.apache.ignite.internal.processors.cache.transactions.TxSavepointNearCacheVisibilityTest;
5152
import org.apache.ignite.internal.processors.cache.transactions.TxSavepointParameterizedTest;
5253
import org.apache.ignite.internal.processors.cache.transactions.TxWithKeyContentionSelfTest;
5354
import org.apache.ignite.testframework.GridTestUtils;
@@ -113,6 +114,7 @@ public static List<Class<?>> suite(Collection<Class> ignoredTests) {
113114
GridTestUtils.addTestIfNeeded(suite, DelayedOwningDuringExchangeTest.class, ignoredTests);
114115

115116
GridTestUtils.addTestIfNeeded(suite, TxSavepointItTest.class, ignoredTests);
117+
GridTestUtils.addTestIfNeeded(suite, TxSavepointNearCacheVisibilityTest.class, ignoredTests);
116118
GridTestUtils.addTestIfNeeded(suite, TxSavepointParameterizedTest.class, ignoredTests);
117119

118120
return suite;

0 commit comments

Comments
 (0)