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/11/02 08:49:15 UTC
ignite git commit: IGNITE-9841: SQL: Improved tracking of LOST
partitions for queries when persistence is enabled. This closes #5069.
Repository: ignite
Updated Branches:
refs/heads/master de5e577c4 -> 5cf44e8e3
IGNITE-9841: SQL: Improved tracking of LOST partitions for queries when persistence is enabled. This closes #5069.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/5cf44e8e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/5cf44e8e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/5cf44e8e
Branch: refs/heads/master
Commit: 5cf44e8e3af75dcbf45e2ad2b230550b27e87923
Parents: de5e577
Author: Stanislav Lukyanov <st...@gmail.com>
Authored: Fri Nov 2 11:49:00 2018 +0300
Committer: devozerov <vo...@gridgain.com>
Committed: Fri Nov 2 11:49:00 2018 +0300
----------------------------------------------------------------------
.../IgniteCachePartitionLossPolicySelfTest.java | 435 +++++++++++++++++--
.../h2/twostep/GridReduceQueryExecutor.java | 21 +
...ndexingCachePartitionLossPolicySelfTest.java | 87 +---
3 files changed, 428 insertions(+), 115 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/5cf44e8e/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCachePartitionLossPolicySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCachePartitionLossPolicySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCachePartitionLossPolicySelfTest.java
index 1616e8f..caf0829 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCachePartitionLossPolicySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCachePartitionLossPolicySelfTest.java
@@ -24,17 +24,23 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.cache.CacheException;
+import junit.framework.AssertionFailedError;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.PartitionLossPolicy;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.events.CacheRebalancingEvent;
import org.apache.ignite.events.Event;
@@ -81,7 +87,10 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
private final AtomicBoolean delayPartExchange = new AtomicBoolean(false);
/** */
- private final TopologyChanger killSingleNode = new TopologyChanger(false, Arrays.asList(3), Arrays.asList(0, 1, 2, 4),0);
+ private final TopologyChanger killSingleNode = new TopologyChanger(false, Collections.singletonList(3), Arrays.asList(0, 1, 2, 4), 0);
+
+ /** */
+ private boolean isPersistenceEnabled;
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
@@ -103,6 +112,10 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
cfg.setCacheConfiguration(cacheConfiguration());
+ cfg.setDataStorageConfiguration(new DataStorageConfiguration().setDefaultDataRegionConfiguration(
+ new DataRegionConfiguration().setPersistenceEnabled(isPersistenceEnabled)
+ ));
+
return cfg;
}
@@ -122,17 +135,25 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
}
/** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- stopAllGrids();
- }
-
- /** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
super.beforeTest();
delayPartExchange.set(false);
+ partLossPlc = PartitionLossPolicy.IGNORE;
+
backups = 0;
+
+ isPersistenceEnabled = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ cleanPersistenceDir();
+
+ super.afterTest();
}
/**
@@ -147,6 +168,17 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadOnlySafeWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_ONLY_SAFE;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(false, true, killSingleNode);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadOnlyAll() throws Exception {
partLossPlc = PartitionLossPolicy.READ_ONLY_ALL;
@@ -156,6 +188,19 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadOnlyAllWithPersistence() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-10041");
+
+ partLossPlc = PartitionLossPolicy.READ_ONLY_ALL;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(false, false, killSingleNode);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafe() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -165,6 +210,17 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteSafeWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, killSingleNode);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteAll() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_ALL;
@@ -174,6 +230,19 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteAllWithPersistence() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-10041");
+
+ partLossPlc = PartitionLossPolicy.READ_WRITE_ALL;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, false, killSingleNode);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafeAfterKillTwoNodes() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -183,6 +252,17 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteSafeAfterKillTwoNodesWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, new TopologyChanger(false, Arrays.asList(3, 2), Arrays.asList(0, 1, 4), 0));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafeAfterKillTwoNodesWithDelay() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -192,6 +272,17 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteSafeAfterKillTwoNodesWithDelayWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, new TopologyChanger(false, Arrays.asList(3, 2), Arrays.asList(0, 1, 4), 20));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafeWithBackupsAfterKillThreeNodes() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -203,6 +294,21 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteSafeWithBackupsAfterKillThreeNodesWithPersistence() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-10043");
+
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ backups = 1;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, new TopologyChanger(true, Arrays.asList(3, 2, 1), Arrays.asList(0, 4), 0));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafeAfterKillCrd() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -212,6 +318,17 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteSafeAfterKillCrdWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, new TopologyChanger(true, Arrays.asList(3, 0), Arrays.asList(1, 2, 4), 0));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafeWithBackups() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -223,6 +340,19 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
/**
* @throws Exception if failed.
*/
+ public void testReadWriteSafeWithBackupsWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ backups = 1;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, new TopologyChanger(true, Arrays.asList(3, 2), Arrays.asList(0, 1, 4), 0));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
public void testReadWriteSafeWithBackupsAfterKillCrd() throws Exception {
partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
@@ -232,12 +362,81 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
}
/**
- * @param topChanger topology changer.
* @throws Exception if failed.
*/
- public void testIgnore(TopologyChanger topChanger) throws Exception {
+ public void testReadWriteSafeWithBackupsAfterKillCrdWithPersistence() throws Exception {
+ partLossPlc = PartitionLossPolicy.READ_WRITE_SAFE;
+
+ backups = 1;
+
+ isPersistenceEnabled = true;
+
+ checkLostPartition(true, true, new TopologyChanger(true, Arrays.asList(3, 0), Arrays.asList(1, 2, 4), 0));
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testIgnore() throws Exception {
fail("https://issues.apache.org/jira/browse/IGNITE-5078");
+ partLossPlc = PartitionLossPolicy.IGNORE;
+
+ checkIgnore(killSingleNode);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testIgnoreWithPersistence() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-5078");
+
+ fail("https://issues.apache.org/jira/browse/IGNITE-10041");
+
+ partLossPlc = PartitionLossPolicy.IGNORE;
+
+ isPersistenceEnabled = true;
+
+ checkIgnore(killSingleNode);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testIgnoreKillThreeNodes() throws Exception {
+ partLossPlc = PartitionLossPolicy.IGNORE;
+
+ // TODO aliveNodes should include node 4, but it fails due to https://issues.apache.org/jira/browse/IGNITE-5078.
+ // TODO need to add 4 to the aliveNodes after IGNITE-5078 is fixed.
+ // TopologyChanger onlyCrdIsAlive = new TopologyChanger(false, Arrays.asList(1, 2, 3), Arrays.asList(0, 4), 0);
+ TopologyChanger onlyCrdIsAlive = new TopologyChanger(false, Arrays.asList(1, 2, 3), Collections.singletonList(0), 0);
+
+ checkIgnore(onlyCrdIsAlive);
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ public void testIgnoreKillThreeNodesWithPersistence() throws Exception {
+ fail("https://issues.apache.org/jira/browse/IGNITE-10041");
+
+ partLossPlc = PartitionLossPolicy.IGNORE;
+
+ isPersistenceEnabled = true;
+
+ // TODO aliveNodes should include node 4, but it fails due to https://issues.apache.org/jira/browse/IGNITE-5078.
+ // TODO need to add 4 to the aliveNodes after IGNITE-5078 is fixed.
+ // TopologyChanger onlyCrdIsAlive = new TopologyChanger(false, Arrays.asList(1, 2, 3), Arrays.asList(0, 4), 0);
+ TopologyChanger onlyCrdIsAlive = new TopologyChanger(false, Arrays.asList(1, 2, 3), Collections.singletonList(0), 0);
+
+ checkIgnore(onlyCrdIsAlive);
+ }
+
+ /**
+ * @param topChanger topology changer.
+ * @throws Exception if failed.
+ */
+ private void checkIgnore(TopologyChanger topChanger) throws Exception {
topChanger.changeTopology();
for (Ignite ig : G.allGrids()) {
@@ -266,14 +465,14 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
private void checkLostPartition(boolean canWrite, boolean safe, TopologyChanger topChanger) throws Exception {
assert partLossPlc != null;
- int part = topChanger.changeTopology().get(0);
+ List<Integer> lostParts = topChanger.changeTopology();
// Wait for all grids (servers and client) have same topology version
// to make sure that all nodes received map with lost partition.
- GridTestUtils.waitForCondition(() -> {
+ boolean success = GridTestUtils.waitForCondition(() -> {
AffinityTopologyVersion last = null;
for (Ignite ig : G.allGrids()) {
- AffinityTopologyVersion ver = ((IgniteEx) ig).context().cache().context().exchange().readyAffinityVersion();
+ AffinityTopologyVersion ver = ((IgniteEx)ig).context().cache().context().exchange().readyAffinityVersion();
if (last != null && !last.equals(ver))
return false;
@@ -284,35 +483,57 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
return true;
}, 10000);
+ assertTrue("Failed to wait for new topology", success);
+
for (Ignite ig : G.allGrids()) {
info("Checking node: " + ig.cluster().localNode().id());
IgniteCache<Integer, Integer> cache = ig.cache(CACHE_NAME);
- verifyCacheOps(canWrite, safe, part, ig);
+ verifyLostPartitions(ig, lostParts);
- // Check we can read and write to lost partition in recovery mode.
- IgniteCache<Integer, Integer> recoverCache = cache.withPartitionRecover();
+ verifyCacheOps(canWrite, safe, ig);
- for (int lostPart : recoverCache.lostPartitions()) {
- recoverCache.get(lostPart);
- recoverCache.put(lostPart, lostPart);
- }
+ validateQuery(safe, ig);
- // Check that writing in recover mode does not clear partition state.
- verifyCacheOps(canWrite, safe, part, ig);
+ // TODO withPartitionRecover doesn't work with BLT - https://issues.apache.org/jira/browse/IGNITE-10041.
+ if (!isPersistenceEnabled) {
+ // Check we can read and write to lost partition in recovery mode.
+ IgniteCache<Integer, Integer> recoverCache = cache.withPartitionRecover();
+
+ for (int lostPart : recoverCache.lostPartitions()) {
+ recoverCache.get(lostPart);
+ recoverCache.put(lostPart, lostPart);
+ }
- // Validate queries.
- validateQuery(safe, part, ig);
+ // Check that writing in recover mode does not clear partition state.
+ verifyLostPartitions(ig, lostParts);
+
+ verifyCacheOps(canWrite, safe, ig);
+
+ validateQuery(safe, ig);
+ }
}
- // Check that partition state does not change after we start a new node.
- IgniteEx grd = startGrid(3);
+ // Bring all nodes back.
+ for (int i : topChanger.killNodes) {
+ IgniteEx grd = startGrid(i);
- info("Newly started node: " + grd.cluster().localNode().id());
+ info("Newly started node: " + grd.cluster().localNode().id());
- for (Ignite ig : G.allGrids())
- verifyCacheOps(canWrite, safe, part, ig);
+ // Check that partition state does not change after we start each node.
+ // TODO With persistence enabled LOST partitions become OWNING after a node joins back - https://issues.apache.org/jira/browse/IGNITE-10044.
+ if (!isPersistenceEnabled) {
+ for (Ignite ig : G.allGrids()) {
+ verifyCacheOps(canWrite, safe, ig);
+
+ // TODO Query effectively waits for rebalance due to https://issues.apache.org/jira/browse/IGNITE-10057
+ // TODO and after resetLostPartition there is another OWNING copy in the cluster due to https://issues.apache.org/jira/browse/IGNITE-10058.
+ // TODO Uncomment after https://issues.apache.org/jira/browse/IGNITE-10058 is fixed.
+// validateQuery(safe, ig);
+ }
+ }
+ }
ignite(4).resetLostPartitions(Collections.singletonList(CACHE_NAME));
@@ -330,24 +551,40 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
cache.put(i, i);
}
+
+ for (int i = 0; i < parts; i++) {
+ checkQueryPasses(ig, false, i);
+
+ if (shouldExecuteLocalQuery(ig, i))
+ checkQueryPasses(ig, true, i);
+
+ }
+
+ checkQueryPasses(ig, false);
}
}
/**
- *
+ * @param node Node.
+ * @param lostParts Lost partition IDs.
+ */
+ private void verifyLostPartitions(Ignite node, List<Integer> lostParts) {
+ IgniteCache<Integer, Integer> cache = node.cache(CACHE_NAME);
+
+ Set<Integer> actualSortedLostParts = new TreeSet<>(cache.lostPartitions());
+ Set<Integer> expSortedLostParts = new TreeSet<>(lostParts);
+
+ assertEqualsCollections(expSortedLostParts, actualSortedLostParts);
+ }
+
+ /**
* @param canWrite {@code True} if writes are allowed.
* @param safe {@code True} if lost partition should trigger exception.
- * @param part Lost partition ID.
* @param ig Ignite instance.
*/
- private void verifyCacheOps(boolean canWrite, boolean safe, int part, Ignite ig) {
+ private void verifyCacheOps(boolean canWrite, boolean safe, Ignite ig) {
IgniteCache<Integer, Integer> cache = ig.cache(CACHE_NAME);
- Collection<Integer> lost = cache.lostPartitions();
-
- assertTrue("Failed to find expected lost partition [exp=" + part + ", lost=" + lost + ']',
- lost.contains(part));
-
int parts = ig.affinity(CACHE_NAME).partitions();
// Check read.
@@ -395,7 +632,7 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
* @param nodes List of nodes to find partition.
* @return List of partitions that aren't primary or backup for specified nodes.
*/
- protected List<Integer> noPrimaryOrBackupPartition(List<Integer> nodes) {
+ private List<Integer> noPrimaryOrBackupPartition(List<Integer> nodes) {
Affinity<Object> aff = ignite(4).affinity(CACHE_NAME);
List<Integer> parts = new ArrayList<>();
@@ -424,15 +661,125 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
* Validate query execution on a node.
*
* @param safe Safe flag.
- * @param part Partition.
* @param node Node.
*/
- protected void validateQuery(boolean safe, int part, Ignite node) {
+ private void validateQuery(boolean safe, Ignite node) {
+ // Get node lost and remaining partitions.
+ IgniteCache<?, ?> cache = node.cache(CACHE_NAME);
+
+ Collection<Integer> lostParts = cache.lostPartitions();
+
+ int part = cache.lostPartitions().stream().findFirst().orElseThrow(AssertionFailedError::new);
+
+ Integer remainingPart = null;
+
+ for (int i = 0; i < node.affinity(CACHE_NAME).partitions(); i++) {
+ if (lostParts.contains(i))
+ continue;
+
+ remainingPart = i;
+
+ break;
+ }
+
+ assertNotNull("Failed to find a partition that isn't lost", remainingPart);
+
+ // 1. Check query against all partitions.
+ validateQuery0(safe, node);
+
+ // 2. Check query against LOST partition.
+ validateQuery0(safe, node, part);
+
+ // 3. Check query on remaining partition.
+ checkQueryPasses(node, false, remainingPart);
+
+ if (shouldExecuteLocalQuery(node, remainingPart))
+ checkQueryPasses(node, true, remainingPart);
+
+ // 4. Check query over two partitions - normal and LOST.
+ validateQuery0(safe, node, part, remainingPart);
+ }
+
+ /**
+ * Query validation routine.
+ *
+ * @param safe Safe flag.
+ * @param node Node.
+ * @param parts Partitions.
+ */
+ private void validateQuery0(boolean safe, Ignite node, int... parts) {
+ if (safe)
+ checkQueryFails(node, false, parts);
+ else
+ checkQueryPasses(node, false, parts);
+
+ if (shouldExecuteLocalQuery(node, parts)) {
+ if (safe)
+ checkQueryFails(node, true, parts);
+ else
+ checkQueryPasses(node, true, parts);
+ }
+ }
+
+ /**
+ * @return true if the given node is primary for all given partitions.
+ */
+ private boolean shouldExecuteLocalQuery(Ignite node, int... parts) {
+ if (parts == null || parts.length == 0)
+ return false;
+
+ int numOfPrimaryParts = 0;
+
+ for (int nodePrimaryPart : node.affinity(CACHE_NAME).primaryPartitions(node.cluster().localNode())) {
+ for (int part : parts) {
+ if (part == nodePrimaryPart)
+ numOfPrimaryParts++;
+ }
+ }
+
+ return numOfPrimaryParts == parts.length;
+ }
+
+ /**
+ * @param node Node.
+ * @param loc Local flag.
+ * @param parts Partitions.
+ */
+ protected void checkQueryPasses(Ignite node, boolean loc, int... parts) {
+ // Scan queries don't support multiple partitions.
+ if (parts != null && parts.length > 1)
+ return;
+
+ // TODO Local scan queries fail in non-safe modes - https://issues.apache.org/jira/browse/IGNITE-10059.
+ if (loc)
+ return;
+
+ IgniteCache cache = node.cache(CACHE_NAME);
+
+ ScanQuery qry = new ScanQuery();
+
+ if (parts != null && parts.length > 0)
+ qry.setPartition(parts[0]);
+
+ if (loc)
+ qry.setLocal(true);
+
+ cache.query(qry).getAll();
+ }
+
+ /**
+ * @param node Node.
+ * @param loc Local flag.
+ * @param parts Partitions.
+ */
+ protected void checkQueryFails(Ignite node, boolean loc, int... parts) {
+ // TODO Scan queries never fail due to partition loss - https://issues.apache.org/jira/browse/IGNITE-9902.
+ // TODO Need to add an actual check after https://issues.apache.org/jira/browse/IGNITE-9902 is fixed.
// No-op.
}
/** */
- class TopologyChanger {
+ private class TopologyChanger {
/** Flag to delay partition exchange */
private boolean delayExchange;
@@ -451,7 +798,7 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
* @param aliveNodes List of nodes to be alive.
* @param stopDelay Delay between stopping nodes.
*/
- public TopologyChanger(boolean delayExchange, List<Integer> killNodes, List<Integer> aliveNodes,
+ private TopologyChanger(boolean delayExchange, List<Integer> killNodes, List<Integer> aliveNodes,
long stopDelay) {
this.delayExchange = delayExchange;
this.killNodes = killNodes;
@@ -463,9 +810,12 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
* @return Lost partition ID.
* @throws Exception If failed.
*/
- protected List<Integer> changeTopology() throws Exception {
+ private List<Integer> changeTopology() throws Exception {
startGrids(4);
+ if (isPersistenceEnabled)
+ grid(0).cluster().active(true);
+
Affinity<Object> aff = ignite(0).affinity(CACHE_NAME);
for (int i = 0; i < aff.partitions(); i++)
@@ -497,7 +847,6 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
lostMap.add(semaphoreMap);
-
grid(i).events().localListen(new P1<Event>() {
@Override public boolean apply(Event evt) {
assert evt.type() == EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST;
@@ -512,7 +861,6 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
return true;
}
}, EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST);
-
}
if (delayExchange)
@@ -549,5 +897,4 @@ public class IgniteCachePartitionLossPolicySelfTest extends GridCommonAbstractTe
return parts;
}
}
-
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5cf44e8e/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index b30aa2f..87c8ce9 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -47,6 +47,7 @@ import org.apache.ignite.IgniteClientDisconnectedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.PartitionLossPolicy;
import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
@@ -112,6 +113,8 @@ import org.jetbrains.annotations.Nullable;
import static java.util.Collections.singletonList;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SQL_RETRY_TIMEOUT;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_SAFE;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE;
import static org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion.NONE;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.checkActive;
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.mvccEnabled;
@@ -1693,6 +1696,24 @@ public class GridReduceQueryExecutor {
Map<ClusterNode, IntArray> partsMap = null;
Map<ClusterNode, IntArray> qryMap = null;
+ for (int cacheId : cacheIds) {
+ GridCacheContext<?, ?> cctx = cacheContext(cacheId);
+
+ PartitionLossPolicy plc = cctx.config().getPartitionLossPolicy();
+
+ if (plc != READ_ONLY_SAFE && plc != READ_WRITE_SAFE)
+ continue;
+
+ Collection<Integer> lostParts = cctx.topology().lostPartitions();
+
+ for (int part : lostParts) {
+ if (parts == null || Arrays.binarySearch(parts, part) >= 0) {
+ throw new CacheException("Failed to execute query because cache partition has been " +
+ "lost [cacheName=" + cctx.name() + ", part=" + part + ']');
+ }
+ }
+ }
+
if (isPreloadingActive(cacheIds)) {
if (isReplicatedOnly)
nodes = replicatedUnstableDataNodes(cacheIds, qryId);
http://git-wip-us.apache.org/repos/asf/ignite/blob/5cf44e8e/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossPolicySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossPolicySelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossPolicySelfTest.java
index f208599..a31a1c6 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossPolicySelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossPolicySelfTest.java
@@ -23,8 +23,6 @@ import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.processors.cache.distributed.IgniteCachePartitionLossPolicySelfTest;
-import java.util.Collection;
-
/**
* Partition loss policy test with enabled indexing.
*/
@@ -39,80 +37,27 @@ public class IndexingCachePartitionLossPolicySelfTest extends IgniteCachePartiti
}
/** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override protected void validateQuery(boolean safe, int part, Ignite node) {
- // Get node lost and remaining partitions.
- IgniteCache cache = node.cache(CACHE_NAME);
-
- Collection<Integer> lostParts = cache.lostPartitions();
-
- Integer remainingPart = null;
-
- for (int i = 0; i < node.affinity(CACHE_NAME).partitions(); i++) {
- if (lostParts.contains(i))
- continue;
-
- remainingPart = i;
-
- break;
- }
-
- // Determine whether local query should be executed on that node.
- boolean execLocQry = false;
-
- for (int nodePrimaryPart : node.affinity(CACHE_NAME).primaryPartitions(node.cluster().localNode())) {
- if (part == nodePrimaryPart) {
- execLocQry = true;
-
- break;
- }
- }
-
- // 1. Check query against all partitions.
- validateQuery0(safe, node, false);
-
- // TODO: https://issues.apache.org/jira/browse/IGNITE-7039
-// if (execLocQry)
-// validateQuery0(safe, node, true);
-
- // 2. Check query against LOST partition.
- validateQuery0(safe, node, false, part);
+ protected void checkQueryPasses(Ignite node, boolean loc, int... parts) {
+ executeQuery(node, loc, parts);
+ }
- // TODO: https://issues.apache.org/jira/browse/IGNITE-7039
-// if (execLocQry)
-// validateQuery0(safe, node, true, part);
+ /** {@inheritDoc} */
+ protected void checkQueryFails(Ignite node, boolean loc, int... parts) {
+ // TODO: Local queries ignore partition loss, see https://issues.apache.org/jira/browse/IGNITE-7039.
+ if (loc)
+ return;
- // 3. Check query on remaining partition.
- if (remainingPart != null) {
- executeQuery(node, false, remainingPart);
+ try {
+ executeQuery(node, loc, parts);
- // 4. Check query over two partitions - normal and LOST.
- validateQuery0(safe, node, false, part, remainingPart);
+ fail("Exception is not thrown.");
}
- }
-
- /**
- * Query validation routine.
- *
- * @param safe Safe flag.
- * @param node Node.
- * @param loc Local flag.
- * @param parts Partitions.
- */
- private void validateQuery0(boolean safe, Ignite node, boolean loc, int... parts) {
- if (safe) {
- try {
- executeQuery(node, loc, parts);
+ catch (Exception e) {
+ boolean exp = e.getMessage() != null &&
+ e.getMessage().contains("Failed to execute query because cache partition has been lost");
- fail("Exception is not thrown.");
- }
- catch (Exception e) {
- assertTrue(e.getMessage(), e.getMessage() != null &&
- e.getMessage().contains("Failed to execute query because cache partition has been lost"));
- }
- }
- else {
- executeQuery(node, loc, parts);
+ if (!exp)
+ throw e;
}
}