You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2018/10/18 09:19:47 UTC
ignite git commit: IGNITE-9927: Reverted changes to
CacheContinuousQueryOperationFromCallbackTest until flaky failures are fixed.
Repository: ignite
Updated Branches:
refs/heads/master 1b6981000 -> a9f37a2e4
IGNITE-9927: Reverted changes to CacheContinuousQueryOperationFromCallbackTest until flaky failures are fixed.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a9f37a2e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a9f37a2e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a9f37a2e
Branch: refs/heads/master
Commit: a9f37a2e4800df5eefc81345292d9b4a4ab94c67
Parents: 1b69810
Author: devozerov <vo...@gridgain.com>
Authored: Thu Oct 18 12:19:38 2018 +0300
Committer: devozerov <vo...@gridgain.com>
Committed: Thu Oct 18 12:19:38 2018 +0300
----------------------------------------------------------------------
...ontinuousQueryOperationFromCallbackTest.java | 205 ++++---------------
1 file changed, 39 insertions(+), 166 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9f37a2e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
index 3cb13bf..0540b43 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryOperationFromCallbackTest.java
@@ -63,13 +63,10 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
-import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.PRIMARY_SYNC;
-import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
-import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
/**
*
@@ -200,60 +197,6 @@ public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbs
}
/**
- * @throws Exception If failed.
- */
- public void testMvccTxTwoBackupsFilter() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL_SNAPSHOT, FULL_SYNC);
-
- doTest(ccfg, false);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testMvccTxTwoBackupsFilterPrimary() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL_SNAPSHOT, PRIMARY_SYNC);
-
- doTest(ccfg, false);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testMvccTxReplicatedFilter() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 0, TRANSACTIONAL_SNAPSHOT, FULL_SYNC);
-
- doTest(ccfg, false);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testMvccTxTwoBackup() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(PARTITIONED, 2, TRANSACTIONAL_SNAPSHOT, FULL_SYNC);
-
- doTest(ccfg, true);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testMvccTxReplicated() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 2, TRANSACTIONAL_SNAPSHOT, FULL_SYNC);
-
- doTest(ccfg, true);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testMvccTxReplicatedPrimary() throws Exception {
- CacheConfiguration<Object, Object> ccfg = cacheConfiguration(REPLICATED, 2, TRANSACTIONAL_SNAPSHOT, PRIMARY_SYNC);
-
- doTest(ccfg, true);
- }
-
- /**
* @param ccfg Cache configuration.
* @throws Exception If failed.
*/
@@ -309,51 +252,34 @@ public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbs
QueryTestKey key = new QueryTestKey(rnd.nextInt(KEYS));
- boolean startTx = cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() !=
- ATOMIC && rnd.nextBoolean();
+ boolean startTx = cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() ==
+ TRANSACTIONAL && rnd.nextBoolean();
Transaction tx = null;
- boolean committed = false;
-
- while (!committed && !Thread.currentThread().isInterrupted()) {
- try {
- if (startTx)
- tx = cache.unwrap(Ignite.class).transactions().txStart(PESSIMISTIC, REPEATABLE_READ);
+ if (startTx)
+ tx = cache.unwrap(Ignite.class).transactions().txStart();
- if ((cache.get(key) == null) || rnd.nextBoolean())
- cache.invoke(key, new IncrementTestEntryProcessor());
- else {
- QueryTestValue val;
- QueryTestValue newVal;
+ try {
+ if ((cache.get(key) == null) || rnd.nextBoolean())
+ cache.invoke(key, new IncrementTestEntryProcessor());
+ else {
+ QueryTestValue val;
+ QueryTestValue newVal;
- do {
- val = cache.get(key);
+ do {
+ val = cache.get(key);
- newVal = val == null ?
- new QueryTestValue(0) : new QueryTestValue(val.val1 + 1);
- }
- while (!cache.replace(key, val, newVal));
+ newVal = val == null ?
+ new QueryTestValue(0) : new QueryTestValue(val.val1 + 1);
}
-
- if (tx != null)
- tx.commit();
-
- committed = true;
- }
- catch (Exception e) {
- assertTrue(e.getMessage(), e.getMessage() != null &&
- (e.getMessage().contains("Transaction has been rolled back") ||
- e.getMessage().contains("Cannot serialize transaction due to write conflict")));
-
- // Wait MVCC updates become visible.
- doSleep(50);
- }
- finally {
- if (tx != null)
- tx.close();
+ while (!cache.replace(key, val, newVal));
}
}
+ finally {
+ if (tx != null)
+ tx.commit();
+ }
}
}
}, threadCnt, "put-thread");
@@ -376,7 +302,7 @@ public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbs
@Override public boolean apply() {
return cbCntr.get() >= expCnt;
}
- }, TimeUnit.SECONDS.toMillis(120));
+ }, TimeUnit.SECONDS.toMillis(60));
assertTrue("Failed to wait events [exp=" + expCnt + ", act=" + cbCntr.get() + "]", res);
@@ -393,7 +319,7 @@ public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbs
@Override public boolean apply() {
return filterCbCntr.get() >= expInvkCnt;
}
- }, TimeUnit.SECONDS.toMillis(120));
+ }, TimeUnit.SECONDS.toMillis(60));
assertEquals(expInvkCnt, filterCbCntr.get());
@@ -481,45 +407,17 @@ public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbs
if (e.getKey().compareTo(new QueryTestKey(KEYS)) < 0) {
IgniteCache<QueryTestKey, QueryTestValue> cache = ignite.cache(cacheName);
- boolean committed = false;
- Transaction tx = null;
- boolean startTx = cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() == TRANSACTIONAL_SNAPSHOT;
-
- while (!committed && !Thread.currentThread().isInterrupted()) {
- try {
- if (startTx)
- tx = ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ);
-
- if (ThreadLocalRandom.current().nextBoolean()) {
- Set<QueryTestKey> keys = new LinkedHashSet<>();
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ Set<QueryTestKey> keys = new LinkedHashSet<>();
- for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
- keys.add(new QueryTestKey(key));
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ keys.add(new QueryTestKey(key));
- cache.invokeAll(keys, new IncrementTestEntryProcessor());
- }
- else {
- for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
- cache.invoke(new QueryTestKey(key), new IncrementTestEntryProcessor());
- }
-
- if (tx != null)
- tx.commit();
-
- committed = true;
- }
- catch (Exception ex) {
- assertTrue(ex.getMessage(), ex.getMessage() != null &&
- (ex.getMessage().contains("Transaction has been rolled back") ||
- ex.getMessage().contains("Cannot serialize transaction due to write conflict")));
-
- // Wait MVCC updates become visible.
- doSleep(50);
- }
- finally {
- if (tx != null)
- tx.close();
- }
+ cache.invokeAll(keys, new IncrementTestEntryProcessor());
+ }
+ else {
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ cache.invoke(new QueryTestKey(key), new IncrementTestEntryProcessor());
}
filterCbCntr.incrementAndGet();
@@ -579,42 +477,17 @@ public class CacheContinuousQueryOperationFromCallbackTest extends GridCommonAbs
cntr.incrementAndGet();
if (cache != null) {
- boolean committed = false;
- Transaction tx = null;
- boolean startTx = cache.getConfiguration(CacheConfiguration.class).getAtomicityMode() == TRANSACTIONAL_SNAPSHOT;
-
- while (!committed && !Thread.currentThread().isInterrupted()) {
- try {
- if (startTx)
- tx = cache.unwrap(Ignite.class).transactions().txStart(PESSIMISTIC, REPEATABLE_READ);
-
- if (ThreadLocalRandom.current().nextBoolean()) {
- Set<QueryTestKey> keys = new LinkedHashSet<>();
-
- for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
- keys.add(new QueryTestKey(key));
-
- cache.invokeAll(keys, new IncrementTestEntryProcessor());
- }
- else {
- for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
- cache.invoke(new QueryTestKey(key), new IncrementTestEntryProcessor());
- }
+ if (ThreadLocalRandom.current().nextBoolean()) {
+ Set<QueryTestKey> keys = new LinkedHashSet<>();
- if (tx != null)
- tx.commit();
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ keys.add(new QueryTestKey(key));
- committed = true;
- }
- catch (Exception ex) {
- assertTrue(ex.getMessage(), ex.getMessage() != null &&
- (ex.getMessage().contains("Transaction has been rolled back") ||
- ex.getMessage().contains("Cannot serialize transaction due to write conflict")));
- }
- finally {
- if (tx != null)
- tx.close();
- }
+ cache.invokeAll(keys, new IncrementTestEntryProcessor());
+ }
+ else {
+ for (int key = KEYS; key < KEYS + KEYS_FROM_CALLBACK; key++)
+ cache.invoke(new QueryTestKey(key), new IncrementTestEntryProcessor());
}
}
}