You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/06/15 06:23:47 UTC
[1/4] ignite git commit: ignite-5272 Do not use synchronous custom
discovery event for client cache start/close
Repository: ignite
Updated Branches:
refs/heads/master 3fd9e0419 -> 4026ddcc2
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheGroupsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheGroupsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheGroupsTest.java
index 06f1922..0787535 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheGroupsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheGroupsTest.java
@@ -49,8 +49,8 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheGroupsTest
commitFailed(tx);
}
- assertEmpty(CACHE_NAME_1); // rolled back
- assertEmpty(CACHE_NAME_3); // rolled back
+ assertEmpty(CACHE_NAME_1); // Rolled back.
+ assertEmpty(CACHE_NAME_3); // Rolled back.
try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putInvalid(CACHE_NAME_1);
@@ -65,7 +65,7 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheGroupsTest
putInvalid(CACHE_NAME_1);
}
- assertEmpty(DEFAULT_CACHE_NAME); // rolled back
+ assertEmpty(DEFAULT_CACHE_NAME); // Rolled back.
startGrid(1);
@@ -94,7 +94,7 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheGroupsTest
putInvalid(CACHE_NAME_1);
}
- assertEmpty(CACHE_NAME_3); // rolled back
+ assertEmpty(CACHE_NAME_3); // Rolled back.
try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(CACHE_NAME_1);
@@ -102,8 +102,8 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheGroupsTest
commitFailed(tx);
}
- assertEmpty(CACHE_NAME_1); // rolled back
- assertEmpty(CACHE_NAME_3); // rolled back
+ assertEmpty(CACHE_NAME_1); // Rolled back.
+ assertEmpty(CACHE_NAME_3); // Rolled back.
try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(DEFAULT_CACHE_NAME);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheTest.java
index 7949c50..781d88d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractTxCacheTest.java
@@ -20,10 +20,11 @@ package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.transactions.Transaction;
-import org.apache.ignite.transactions.TransactionConcurrency;
-import org.apache.ignite.transactions.TransactionIsolation;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
/**
* Topology validator test
@@ -41,17 +42,16 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
/** {@inheritDoc} */
@Override public void testTopologyValidator() throws Exception {
-
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(CACHE_NAME_1);
commitFailed(tx);
}
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putInvalid(CACHE_NAME_1);
}
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(CACHE_NAME_1);
putValid(DEFAULT_CACHE_NAME);
putValid(CACHE_NAME_2);
@@ -62,7 +62,7 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
assertEmpty(CACHE_NAME_1); // rolled back
assertEmpty(CACHE_NAME_2); // rolled back
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putValid(DEFAULT_CACHE_NAME);
putInvalid(CACHE_NAME_1);
}
@@ -72,14 +72,14 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
startGrid(1);
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(CACHE_NAME_1);
tx.commit();
}
remove(CACHE_NAME_1);
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putValid(CACHE_NAME_1);
tx.commit();
}
@@ -88,7 +88,7 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
startGrid(2);
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putValid(DEFAULT_CACHE_NAME);
putInvalid(CACHE_NAME_1);
}
@@ -96,16 +96,16 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
assertEmpty(DEFAULT_CACHE_NAME); // rolled back
assertEmpty(CACHE_NAME_1); // rolled back
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(CACHE_NAME_1);
commitFailed(tx);
}
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putInvalid(CACHE_NAME_1);
}
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.OPTIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
putValid(DEFAULT_CACHE_NAME);
putValid(CACHE_NAME_2);
tx.commit();
@@ -114,7 +114,7 @@ public abstract class IgniteTopologyValidatorAbstractTxCacheTest extends IgniteT
remove(DEFAULT_CACHE_NAME);
remove(CACHE_NAME_2);
- try (Transaction tx = grid(0).transactions().txStart(TransactionConcurrency.PESSIMISTIC, TransactionIsolation.REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
putValid(DEFAULT_CACHE_NAME);
putValid(CACHE_NAME_2);
tx.commit();
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
index 20cef30..db3f5cd 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
@@ -476,7 +476,7 @@ public class CacheLateAffinityAssignmentTest extends GridCommonAbstractTest {
ignite0.cache(CACHE_NAME1);
- checkAffinity(nodes, topVer(topVer, 2), true);
+ checkAffinity(nodes, topVer(topVer, 1), true);
topVer++;
@@ -486,7 +486,7 @@ public class CacheLateAffinityAssignmentTest extends GridCommonAbstractTest {
ignite0.cache(CACHE_NAME1).close();
- checkAffinity(nodes, topVer(topVer, 2), true);
+ checkAffinity(nodes, topVer(topVer, 1), true);
}
}
@@ -832,11 +832,11 @@ public class CacheLateAffinityAssignmentTest extends GridCommonAbstractTest {
assertNotNull(client.cache(CACHE_NAME1));
- checkAffinity(4, topVer(4, 1), false);
+ checkAffinity(4, topVer(4, 0), false);
spi.stopBlock();
- checkAffinity(4, topVer(4, 2), true);
+ checkAffinity(4, topVer(4, 1), true);
}
/**
@@ -1240,11 +1240,11 @@ public class CacheLateAffinityAssignmentTest extends GridCommonAbstractTest {
IgniteCache cache = client.cache(CACHE_NAME1);
- checkAffinity(2, topVer(2, 1), true);
+ checkAffinity(2, topVer(2, 0), true);
cache.close();
- checkAffinity(2, topVer(2, 2), true);
+ checkAffinity(2, topVer(2, 0), true);
}
/**
@@ -1274,11 +1274,11 @@ public class CacheLateAffinityAssignmentTest extends GridCommonAbstractTest {
client.cache(CACHE_NAME2);
- checkAffinity(4, topVer(4, 2), true);
+ checkAffinity(4, topVer(4, 1), true);
client.destroyCache(CACHE_NAME2);
- checkAffinity(4, topVer(4, 3), true);
+ checkAffinity(4, topVer(4, 2), true);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodePartitionsExchangeTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodePartitionsExchangeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodePartitionsExchangeTest.java
index 0885b1a..1f850b2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodePartitionsExchangeTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodePartitionsExchangeTest.java
@@ -530,26 +530,31 @@ public class IgniteCacheClientNodePartitionsExchangeTest extends GridCommonAbstr
else
ignite2.cache(CACHE_NAME1);
- waitForTopologyUpdate(3, new AffinityTopologyVersion(3, ++minorVer));
-
GridCacheAdapter cache = ((IgniteKernal)ignite2).context().cache().context().cache().internalCache(CACHE_NAME1);
assertNotNull(cache);
assertEquals(nearCache, cache.context().isNear());
assertEquals(0, spi0.partitionsSingleMessages());
- assertEquals(1, spi0.partitionsFullMessages());
+ assertEquals(0, spi0.partitionsFullMessages());
assertEquals(0, spi1.partitionsSingleMessages());
assertEquals(0, spi1.partitionsFullMessages());
- assertEquals(1, spi2.partitionsSingleMessages());
+ assertEquals(0, spi2.partitionsSingleMessages());
assertEquals(0, spi2.partitionsFullMessages());
- ClusterNode clientNode = ((IgniteKernal)ignite2).localNode();
+ final ClusterNode clientNode = ((IgniteKernal)ignite2).localNode();
for (Ignite ignite : Ignition.allGrids()) {
- GridDiscoveryManager disco = ((IgniteKernal)ignite).context().discovery();
+ final GridDiscoveryManager disco = ((IgniteKernal)ignite).context().discovery();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return disco.cacheNode(clientNode, CACHE_NAME1);
+ }
+ }, 5000);
assertTrue(disco.cacheNode(clientNode, CACHE_NAME1));
+
assertFalse(disco.cacheAffinityNode(clientNode, CACHE_NAME1));
assertEquals(nearCache, disco.cacheNearNode(clientNode, CACHE_NAME1));
}
@@ -565,8 +570,6 @@ public class IgniteCacheClientNodePartitionsExchangeTest extends GridCommonAbstr
assertNull(((IgniteKernal)ignite2).context().cache().context().cache().internalCache(CACHE_NAME1));
- waitForTopologyUpdate(3, new AffinityTopologyVersion(3, ++minorVer));
-
assertEquals(0, spi0.partitionsSingleMessages());
assertEquals(0, spi0.partitionsFullMessages());
assertEquals(0, spi1.partitionsSingleMessages());
@@ -577,9 +580,7 @@ public class IgniteCacheClientNodePartitionsExchangeTest extends GridCommonAbstr
final String CACHE_NAME2 = "cache2";
- ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
-
- ccfg.setName(CACHE_NAME2);
+ ccfg = new CacheConfiguration(CACHE_NAME2);
log.info("Create new cache: " + CACHE_NAME2);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java
index aea7a2f..58eceaa 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java
@@ -115,6 +115,7 @@ import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
@@ -193,6 +194,7 @@ public abstract class GridAbstractTest extends TestCase {
System.setProperty(IgniteSystemProperties.IGNITE_ATOMIC_CACHE_DELETE_HISTORY_SIZE, "10000");
System.setProperty(IgniteSystemProperties.IGNITE_UPDATE_NOTIFIER, "false");
System.setProperty(IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY, "1");
+ System.setProperty(IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT, "1000");
if (BINARY_MARSHALLER)
GridTestProperties.setProperty(GridTestProperties.MARSH_CLASS_NAME, BinaryMarshaller.class.getName());
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
index 43f6b13..c69cac6 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
@@ -42,7 +42,9 @@ import org.apache.ignite.internal.processors.cache.IgniteCacheEntryProcessorNode
import org.apache.ignite.internal.processors.cache.IgniteCacheIncrementTxTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheNoSyncForGetTest;
import org.apache.ignite.internal.processors.cache.IgniteCachePartitionMapUpdateTest;
+import org.apache.ignite.internal.processors.cache.IgniteClientCacheStartFailoverTest;
import org.apache.ignite.internal.processors.cache.IgniteDynamicCacheAndNodeStop;
+import org.apache.ignite.internal.processors.cache.IgniteNearClientCacheCloseTest;
import org.apache.ignite.internal.processors.cache.IgniteOnePhaseCommitInvokeTest;
import org.apache.ignite.internal.processors.cache.IgniteOnePhaseCommitNearReadersTest;
import org.apache.ignite.internal.processors.cache.MemoryPolicyConfigValidationTest;
@@ -272,6 +274,8 @@ public class IgniteCacheTestSuite2 extends TestSuite {
suite.addTest(new TestSuite(IgniteCacheNoSyncForGetTest.class));
suite.addTest(new TestSuite(IgniteOnePhaseCommitNearReadersTest.class));
+ suite.addTest(new TestSuite(IgniteNearClientCacheCloseTest.class));
+ suite.addTest(new TestSuite(IgniteClientCacheStartFailoverTest.class));
return suite;
}
[3/4] ignite git commit: ignite-5272 Do not use synchronous custom
discovery event for client cache start/close
Posted by sb...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index 37e495e..09eea27 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -36,6 +36,7 @@ import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
+import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.managers.communication.GridMessageListener;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -301,52 +302,151 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
*/
@SuppressWarnings("unchecked")
private void handleMessage(UUID nodeId, GridCacheMessage cacheMsg, MessageHandlers msgHandlers) {
- int msgIdx = cacheMsg.lookupIndex();
+ Lock lock = rw.readLock();
- IgniteBiInClosure<UUID, GridCacheMessage> c = null;
+ lock.lock();
- if (msgIdx >= 0) {
- Map<Integer, IgniteBiInClosure[]> idxClsHandlers0 = msgHandlers.idxClsHandlers;
+ try {
+ int msgIdx = cacheMsg.lookupIndex();
- IgniteBiInClosure[] cacheClsHandlers = idxClsHandlers0.get(cacheMsg.handlerId());
+ IgniteBiInClosure<UUID, GridCacheMessage> c = null;
- if (cacheClsHandlers != null)
- c = cacheClsHandlers[msgIdx];
- }
+ if (msgIdx >= 0) {
+ Map<Integer, IgniteBiInClosure[]> idxClsHandlers0 = msgHandlers.idxClsHandlers;
- if (c == null)
- c = msgHandlers.clsHandlers.get(new ListenerKey(cacheMsg.handlerId(), cacheMsg.getClass()));
+ IgniteBiInClosure[] cacheClsHandlers = idxClsHandlers0.get(cacheMsg.handlerId());
- if (c == null) {
- IgniteLogger log = cacheMsg.messageLogger(cctx);
+ if (cacheClsHandlers != null)
+ c = cacheClsHandlers[msgIdx];
+ }
- StringBuilder msg0 = new StringBuilder("Received message without registered handler (will ignore) [");
+ if (c == null)
+ c = msgHandlers.clsHandlers.get(new ListenerKey(cacheMsg.handlerId(), cacheMsg.getClass()));
- appendMessageInfo(cacheMsg, nodeId, msg0);
+ if (c == null) {
+ if (processMissedHandler(nodeId, cacheMsg))
+ return;
- msg0.append(", locTopVer=").append(cctx.exchange().readyAffinityVersion()).
- append(", msgTopVer=").append(cacheMsg.topologyVersion()).
- append(", desc=").append(descriptorForMessage(cacheMsg)).
- append(']');
+ IgniteLogger log = cacheMsg.messageLogger(cctx);
- msg0.append(U.nl()).append("Registered listeners:");
+ StringBuilder msg0 = new StringBuilder("Received message without registered handler (will ignore) [");
- Map<Integer, IgniteBiInClosure[]> idxClsHandlers0 = msgHandlers.idxClsHandlers;
+ appendMessageInfo(cacheMsg, nodeId, msg0);
- for (Map.Entry<Integer, IgniteBiInClosure[]> e : idxClsHandlers0.entrySet())
- msg0.append(U.nl()).append(e.getKey()).append("=").append(Arrays.toString(e.getValue()));
+ msg0.append(", locTopVer=").append(cctx.exchange().readyAffinityVersion()).
+ append(", msgTopVer=").append(cacheMsg.topologyVersion()).
+ append(", desc=").append(descriptorForMessage(cacheMsg)).
+ append(']');
- if (cctx.kernalContext().isStopping()) {
- if (log.isDebugEnabled())
- log.debug(msg0.toString());
+ msg0.append(U.nl()).append("Registered listeners:");
+
+ Map<Integer, IgniteBiInClosure[]> idxClsHandlers0 = msgHandlers.idxClsHandlers;
+
+ for (Map.Entry<Integer, IgniteBiInClosure[]> e : idxClsHandlers0.entrySet())
+ msg0.append(U.nl()).append(e.getKey()).append("=").append(Arrays.toString(e.getValue()));
+
+ if (cctx.kernalContext().isStopping()) {
+ if (log.isDebugEnabled())
+ log.debug(msg0.toString());
+ }
+ else
+ U.error(log, msg0.toString());
+
+ return;
}
- else
- U.error(log, msg0.toString());
- return;
+ onMessage0(nodeId, cacheMsg, c);
+ }
+ finally {
+ lock.unlock();
}
+ }
+
+ /**
+ * @param nodeId Node ID.
+ * @param cacheMsg Message.
+ * @return {@code True} if message processed.
+ */
+ private boolean processMissedHandler(UUID nodeId, GridCacheMessage cacheMsg) {
+ // It is possible to receive reader update after client near cache was closed.
+ if (cacheMsg instanceof GridDhtAtomicAbstractUpdateRequest) {
+ GridDhtAtomicAbstractUpdateRequest req = (GridDhtAtomicAbstractUpdateRequest)cacheMsg;
+
+ if (req.nearSize() > 0) {
+ List<KeyCacheObject> nearEvicted = new ArrayList<>(req.nearSize());
+
+ for (int i = 0; i < req.nearSize(); i++)
+ nearEvicted.add(req.nearKey(i));
+
+ GridDhtAtomicUpdateResponse dhtRes = new GridDhtAtomicUpdateResponse(req.cacheId(),
+ req.partition(),
+ req.futureId(),
+ false);
+
+ dhtRes.nearEvicted(nearEvicted);
+
+ sendMessageForMissedHandler(cacheMsg,
+ nodeId,
+ dhtRes,
+ nodeId,
+ GridIoPolicy.SYSTEM_POOL);
+
+ if (req.nearNodeId() != null) {
+ GridDhtAtomicNearResponse nearRes = new GridDhtAtomicNearResponse(req.cacheId(),
+ req.partition(),
+ req.nearFutureId(),
+ nodeId,
+ req.flags());
+
+ sendMessageForMissedHandler(cacheMsg,
+ nodeId,
+ nearRes,
+ req.nearNodeId(),
+ GridIoPolicy.SYSTEM_POOL);
+ }
+
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ /**
+ * @param origMsg Message without handler.
+ * @param origMsgNode Node sent {@code origMsg}.
+ * @param nodeId Target node ID.
+ * @param msg Response.
+ * @param plc Policy.
+ */
+ private void sendMessageForMissedHandler(
+ GridCacheMessage origMsg,
+ UUID origMsgNode,
+ GridCacheMessage msg,
+ UUID nodeId,
+ byte plc) {
+ IgniteLogger log = msg.messageLogger(cctx);
- onMessage0(nodeId, cacheMsg, c);
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug("Received message without registered handler, " +
+ "send response [locTopVer=" + cctx.exchange().readyAffinityVersion() +
+ ", msgTopVer=" + origMsg.topologyVersion() +
+ ", node=" + origMsgNode +
+ ", msg=" + origMsg +
+ ", resNode=" + nodeId +
+ ", res=" + msg + ']');
+ }
+
+ send(nodeId, msg, plc);
+ }
+ catch (ClusterTopologyCheckedException e) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to send response, node left [nodeId=" + nodeId + ", msg=" + msg + ']');
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send response [nodeId=" + nodeId + ", msg=" + msg + ", err=" + e + ']');
+ }
}
/** {@inheritDoc} */
@@ -359,17 +459,10 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
cctx.gridIO().addMessageListener(TOPIC_CACHE, lsnr);
}
- /** {@inheritDoc} */
- @SuppressWarnings("BusyWait")
- @Override protected void onKernalStop0(boolean cancel) {
- cctx.gridIO().removeMessageListener(TOPIC_CACHE);
-
- for (Object ordTopic : cacheHandlers.orderedHandlers.keySet())
- cctx.gridIO().removeMessageListener(ordTopic);
-
- for (Object ordTopic : grpHandlers.orderedHandlers.keySet())
- cctx.gridIO().removeMessageListener(ordTopic);
-
+ /**
+ *
+ */
+ public void writeLock() {
boolean interrupted = false;
// Busy wait is intentional.
@@ -389,6 +482,27 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (interrupted)
Thread.currentThread().interrupt();
+ }
+
+ /**
+ *
+ */
+ public void writeUnlock() {
+ rw.writeLock().unlock();
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("BusyWait")
+ @Override protected void onKernalStop0(boolean cancel) {
+ cctx.gridIO().removeMessageListener(TOPIC_CACHE);
+
+ for (Object ordTopic : cacheHandlers.orderedHandlers.keySet())
+ cctx.gridIO().removeMessageListener(ordTopic);
+
+ for (Object ordTopic : grpHandlers.orderedHandlers.keySet())
+ cctx.gridIO().removeMessageListener(ordTopic);
+
+ writeLock();
try {
stopping = true;
@@ -406,10 +520,6 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
@SuppressWarnings({"unchecked", "ConstantConditions", "ThrowableResultOfMethodCallIgnored"})
private void onMessage0(final UUID nodeId, final GridCacheMessage cacheMsg,
final IgniteBiInClosure<UUID, GridCacheMessage> c) {
- Lock lock = rw.readLock();
-
- lock.lock();
-
try {
if (stopping) {
if (log.isDebugEnabled())
@@ -438,8 +548,6 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
finally {
if (depEnabled)
cctx.deploy().ignoreOwnership(false);
-
- lock.unlock();
}
}
@@ -1527,9 +1635,18 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
if (log.isDebugEnabled())
log.debug("Received cache ordered message [nodeId=" + nodeId + ", msg=" + msg + ']');
- final GridCacheMessage cacheMsg = (GridCacheMessage)msg;
+ Lock lock = rw.readLock();
- onMessage0(nodeId, cacheMsg, c);
+ lock.lock();
+
+ try {
+ GridCacheMessage cacheMsg = (GridCacheMessage)msg;
+
+ onMessage0(nodeId, cacheMsg, c);
+ }
+ finally {
+ lock.unlock();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index fdf8a2f..901667f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -337,6 +337,15 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
}
/**
+ * @param task Task to run in exchange worker thread.
+ */
+ public void addCustomTask(CachePartitionExchangeWorkerTask task) {
+ assert task != null;
+
+ exchWorker.addCustomTask(task);
+ }
+
+ /**
* @return Reconnect partition exchange future.
*/
public IgniteInternalFuture<?> reconnectExchangeFuture() {
@@ -1731,7 +1740,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
cctx.cache().processCustomExchangeTask(task);
}
catch (Exception e) {
- U.warn(log, "Failed to process custom exchange task: " + task, e);
+ U.error(log, "Failed to process custom exchange task: " + task, e);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index ad83b14..96b45df 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -184,7 +184,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
private final Map<String, GridCacheAdapter> stoppedCaches = new ConcurrentHashMap<>();
/** Map of proxies. */
- private final Map<String, IgniteCacheProxy<?, ?>> jCacheProxies;
+ private final ConcurrentHashMap<String, IgniteCacheProxy<?, ?>> jCacheProxies;
/** Caches stop sequence. */
private final Deque<String> stopSeq;
@@ -363,6 +363,11 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (msg0.exchange())
return new SchemaExchangeWorkerTask(msg0);
}
+ else if (msg instanceof ClientCacheChangeDummyDiscoveryMessage) {
+ ClientCacheChangeDummyDiscoveryMessage msg0 = (ClientCacheChangeDummyDiscoveryMessage)msg;
+
+ return msg0;
+ }
return null;
}
@@ -372,7 +377,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
*
* @param task Task.
*/
- public void processCustomExchangeTask(CachePartitionExchangeWorkerTask task) {
+ void processCustomExchangeTask(CachePartitionExchangeWorkerTask task) {
if (task instanceof SchemaExchangeWorkerTask) {
SchemaAbstractDiscoveryMessage msg = ((SchemaExchangeWorkerTask)task).message();
@@ -389,6 +394,16 @@ public class GridCacheProcessor extends GridProcessorAdapter {
ctx.query().onNodeLeave(task0.node());
}
+ else if (task instanceof ClientCacheChangeDummyDiscoveryMessage) {
+ ClientCacheChangeDummyDiscoveryMessage task0 = (ClientCacheChangeDummyDiscoveryMessage)task;
+
+ sharedCtx.affinity().processClientCachesChanges(task0);
+ }
+ else if (task instanceof ClientCacheUpdateTimeout) {
+ ClientCacheUpdateTimeout task0 = (ClientCacheUpdateTimeout)task;
+
+ sharedCtx.affinity().sendClientCacheChangesMessage(task0);
+ }
else
U.warn(log, "Unsupported custom exchange task: " + task);
}
@@ -1998,21 +2013,18 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * @param req Stop request.
+ * @param cacheName Cache name.
+ * @param stop {@code True} for stop cache, {@code false} for close cache.
*/
- void blockGateway(DynamicCacheChangeRequest req) {
- assert req.stop() || req.close();
-
- if (req.stop() || (req.close() && req.initiatingNodeId().equals(ctx.localNodeId()))) {
- // Break the proxy before exchange future is done.
- IgniteCacheProxy<?, ?> proxy = jCacheProxies.get(req.cacheName());
+ void blockGateway(String cacheName, boolean stop) {
+ // Break the proxy before exchange future is done.
+ IgniteCacheProxy<?, ?> proxy = jCacheProxies.get(cacheName);
- if (proxy != null) {
- if (req.stop())
- proxy.gate().stopped();
- else
- proxy.closeProxy();
- }
+ if (proxy != null) {
+ if (stop)
+ proxy.gate().stopped();
+ else
+ proxy.closeProxy();
}
}
@@ -2030,22 +2042,21 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * @param req Stop request.
+ * @param cacheName Cache name.
+ * @param destroy Cache destroy flag.
* @return Cache group for stopped cache.
*/
- private CacheGroupContext prepareCacheStop(DynamicCacheChangeRequest req, boolean forceClose) {
- assert req.stop() || req.close() || forceClose : req;
-
- GridCacheAdapter<?, ?> cache = caches.remove(req.cacheName());
+ private CacheGroupContext prepareCacheStop(String cacheName, boolean destroy) {
+ GridCacheAdapter<?, ?> cache = caches.remove(cacheName);
if (cache != null) {
GridCacheContext<?, ?> ctx = cache.context();
sharedCtx.removeCacheContext(ctx);
- onKernalStop(cache, req.destroy());
+ onKernalStop(cache, destroy);
- stopCache(cache, true, req.destroy());
+ stopCache(cache, true, destroy);
return ctx.group();
}
@@ -2054,24 +2065,91 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Closes cache even if it's not fully initialized (e.g. fail on cache init stage).
- *
- * @param topVer Completed topology version.
- * @param act Exchange action.
- * @param err Error.
+ * @param startTopVer Cache start version.
+ * @param err Cache start error if any.
*/
- void forceCloseCache(
- AffinityTopologyVersion topVer,
- final ExchangeActions.ActionData act,
- Throwable err
- ) {
- ExchangeActions actions = new ExchangeActions(){
- @Override List<ActionData> closeRequests(UUID nodeId) {
- return Collections.singletonList(act);
+ void initCacheProxies(AffinityTopologyVersion startTopVer, @Nullable Throwable err) {
+ for (GridCacheAdapter<?, ?> cache : caches.values()) {
+ GridCacheContext<?, ?> cacheCtx = cache.context();
+
+ if (cacheCtx.startTopologyVersion().equals(startTopVer) && !jCacheProxies.containsKey(cacheCtx.name())) {
+ jCacheProxies.putIfAbsent(cacheCtx.name(), new IgniteCacheProxy(cache.context(), cache, null, false));
+
+ if (cacheCtx.preloader() != null)
+ cacheCtx.preloader().onInitialExchangeComplete(err);
+ }
+ }
+ }
+
+ /**
+ * @param cachesToClose Caches to close.
+ * @param retClientCaches {@code True} if return IDs of closed client caches.
+ * @return Closed client caches' IDs.
+ */
+ Set<Integer> closeCaches(Set<String> cachesToClose, boolean retClientCaches) {
+ Set<Integer> ids = null;
+
+ boolean locked = false;
+
+ try {
+ for (String cacheName : cachesToClose) {
+ blockGateway(cacheName, false);
+
+ GridCacheContext ctx = sharedCtx.cacheContext(CU.cacheId(cacheName));
+
+ if (ctx == null)
+ continue;
+
+ if (retClientCaches && !ctx.affinityNode()) {
+ if (ids == null)
+ ids = U.newHashSet(cachesToClose.size());
+
+ ids.add(ctx.cacheId());
+ }
+
+ if (!ctx.affinityNode() && !locked) {
+ // Do not close client cache while requests processing is in progress.
+ sharedCtx.io().writeLock();
+
+ locked = true;
+ }
+
+ if (!ctx.affinityNode() && ctx.transactional())
+ sharedCtx.tm().rollbackTransactionsForCache(ctx.cacheId());
+
+ closeCache(ctx, false);
}
- };
- onExchangeDone(topVer, actions, err, true);
+ return ids;
+ }
+ finally {
+ if (locked)
+ sharedCtx.io().writeUnlock();
+ }
+ }
+
+ /**
+ * @param cctx Cache context.
+ * @param destroy Destroy flag.
+ */
+ private void closeCache(GridCacheContext cctx, boolean destroy) {
+ if (cctx.affinityNode()) {
+ GridCacheAdapter<?, ?> cache = caches.get(cctx.name());
+
+ assert cache != null : cctx.name();
+
+ jCacheProxies.put(cctx.name(), new IgniteCacheProxy(cache.context(), cache, null, false));
+ }
+ else {
+ jCacheProxies.remove(cctx.name());
+
+ cctx.gate().onStopped();
+
+ CacheGroupContext grp = prepareCacheStop(cctx.name(), destroy);
+
+ if (grp != null && !grp.hasCaches())
+ stopCacheGroup(grp.groupId());
+ }
}
/**
@@ -2085,58 +2163,19 @@ public class GridCacheProcessor extends GridProcessorAdapter {
public void onExchangeDone(
AffinityTopologyVersion topVer,
@Nullable ExchangeActions exchActions,
- Throwable err,
- boolean forceClose
+ Throwable err
) {
- for (GridCacheAdapter<?, ?> cache : caches.values()) {
- GridCacheContext<?, ?> cacheCtx = cache.context();
-
- if (cacheCtx.startTopologyVersion().equals(topVer)) {
- jCacheProxies.put(cacheCtx.name(), new IgniteCacheProxy(cache.context(), cache, null, false));
+ initCacheProxies(topVer, err);
- if (cacheCtx.preloader() != null)
- cacheCtx.preloader().onInitialExchangeComplete(err);
- }
- }
-
- if (exchActions != null && (err == null || forceClose)) {
+ if (exchActions != null && err == null) {
for (ExchangeActions.ActionData action : exchActions.cacheStopRequests()) {
stopGateway(action.request());
- prepareCacheStop(action.request(), forceClose);
+ prepareCacheStop(action.request().cacheName(), action.request().destroy());
}
for (CacheGroupDescriptor grpDesc : exchActions.cacheGroupsToStop())
stopCacheGroup(grpDesc.groupId());
-
- for (ExchangeActions.ActionData req : exchActions.closeRequests(ctx.localNodeId())) {
- String cacheName = req.request().cacheName();
-
- IgniteCacheProxy<?, ?> proxy = jCacheProxies.get(cacheName);
-
- if (proxy != null) {
- if (proxy.context().affinityNode()) {
- GridCacheAdapter<?, ?> cache = caches.get(cacheName);
-
- assert cache != null : cacheName;
-
- jCacheProxies.put(cacheName, new IgniteCacheProxy(cache.context(), cache, null, false));
- }
- else {
- jCacheProxies.remove(cacheName);
-
- proxy.context().gate().onStopped();
-
- CacheGroupContext grp = prepareCacheStop(req.request(), forceClose);
-
- if (grp != null && !grp.hasCaches())
- stopCacheGroup(grp.groupId());
- }
- }
-
- if (forceClose)
- completeCacheStartFuture(req.request(), false, err);
- }
}
}
@@ -2187,6 +2226,17 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
+ * @param reqId Request ID.
+ * @param err Error if any.
+ */
+ void completeClientCacheChangeFuture(UUID reqId, @Nullable Exception err) {
+ DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(reqId);
+
+ if (fut != null)
+ fut.onDone(false, err);
+ }
+
+ /**
* Creates shared context.
*
* @param kernalCtx Kernal context.
@@ -2486,8 +2536,12 @@ public class GridCacheProcessor extends GridProcessorAdapter {
failIfExists,
failIfNotStarted);
- if (req != null)
- return F.first(initiateCacheChanges(F.asList(req), failIfExists));
+ if (req != null) {
+ if (req.clientStartOnly())
+ return startClientCacheChange(F.asMap(req.cacheName(), req), null);
+
+ return F.first(initiateCacheChanges(F.asList(req)));
+ }
else
return new GridFinishedFuture<>();
}
@@ -2497,6 +2551,31 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
+ * @param startReqs Start requests.
+ * @param cachesToClose Cache tp close.
+ * @return Future for cache change operation.
+ */
+ private IgniteInternalFuture<Boolean> startClientCacheChange(
+ @Nullable Map<String, DynamicCacheChangeRequest> startReqs, @Nullable Set<String> cachesToClose) {
+ assert startReqs != null ^ cachesToClose != null;
+
+ DynamicCacheStartFuture fut = new DynamicCacheStartFuture(UUID.randomUUID());
+
+ IgniteInternalFuture old = pendingFuts.put(fut.id, fut);
+
+ assert old == null : old;
+
+ ctx.discovery().clientCacheStartEvent(fut.id, startReqs, cachesToClose);
+
+ IgniteCheckedException err = checkNodeState();
+
+ if (err != null)
+ fut.onDone(err);
+
+ return fut;
+ }
+
+ /**
* Dynamically starts multiple caches.
*
* @param ccfgList Collection of cache configuration.
@@ -2527,7 +2606,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (checkThreadTx)
checkEmptyTransactions();
- List<DynamicCacheChangeRequest> reqList = new ArrayList<>(ccfgList.size());
+ List<DynamicCacheChangeRequest> srvReqs = null;
+ Map<String, DynamicCacheChangeRequest> clientReqs = null;
try {
for (CacheConfiguration ccfg : ccfgList) {
@@ -2541,20 +2621,41 @@ public class GridCacheProcessor extends GridProcessorAdapter {
true
);
- if (req != null)
- reqList.add(req);
+ if (req != null) {
+ if (req.clientStartOnly()) {
+ if (clientReqs == null)
+ clientReqs = U.newLinkedHashMap(ccfgList.size());
+
+ clientReqs.put(req.cacheName(), req);
+ }
+ else {
+ if (srvReqs == null)
+ srvReqs = new ArrayList<>(ccfgList.size());
+
+ srvReqs.add(req);
+ }
+ }
}
}
catch (Exception e) {
return new GridFinishedFuture<>(e);
}
- if (!reqList.isEmpty()) {
+ if (srvReqs != null || clientReqs != null) {
+ if (clientReqs != null && srvReqs == null)
+ return startClientCacheChange(clientReqs, null);
+
GridCompoundFuture<?, ?> compoundFut = new GridCompoundFuture<>();
- for (DynamicCacheStartFuture fut : initiateCacheChanges(reqList, failIfExists))
+ for (DynamicCacheStartFuture fut : initiateCacheChanges(srvReqs))
compoundFut.add((IgniteInternalFuture)fut);
+ if (clientReqs != null) {
+ IgniteInternalFuture<Boolean> clientStartFut = startClientCacheChange(clientReqs, null);
+
+ compoundFut.add((IgniteInternalFuture)clientStartFut);
+ }
+
compoundFut.markInitialized();
return compoundFut;
@@ -2578,7 +2679,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
DynamicCacheChangeRequest req = DynamicCacheChangeRequest.stopRequest(ctx, cacheName, sql, true);
- return F.first(initiateCacheChanges(F.asList(req), false));
+ return F.first(initiateCacheChanges(F.asList(req)));
}
/**
@@ -2600,7 +2701,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
GridCompoundFuture<?, ?> compoundFut = new GridCompoundFuture<>();
- for (DynamicCacheStartFuture fut : initiateCacheChanges(reqs, false))
+ for (DynamicCacheStartFuture fut : initiateCacheChanges(reqs))
compoundFut.add((IgniteInternalFuture)fut);
compoundFut.markInitialized();
@@ -2622,9 +2723,10 @@ public class GridCacheProcessor extends GridProcessorAdapter {
checkEmptyTransactions();
- DynamicCacheChangeRequest t = DynamicCacheChangeRequest.closeRequest(ctx, cacheName);
+ if (proxy.context().isLocal())
+ return dynamicDestroyCache(cacheName, false, true);
- return F.first(initiateCacheChanges(F.asList(t), false));
+ return startClientCacheChange(null, Collections.singleton(cacheName));
}
/**
@@ -2657,7 +2759,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
GridCompoundFuture fut = new GridCompoundFuture();
- for (DynamicCacheStartFuture f : initiateCacheChanges(reqs, false))
+ for (DynamicCacheStartFuture f : initiateCacheChanges(reqs))
fut.add(f);
fut.markInitialized();
@@ -2751,38 +2853,28 @@ public class GridCacheProcessor extends GridProcessorAdapter {
/**
* @param reqs Requests.
- * @param failIfExists Fail if exists flag.
* @return Collection of futures.
*/
@SuppressWarnings("TypeMayBeWeakened")
private Collection<DynamicCacheStartFuture> initiateCacheChanges(
- Collection<DynamicCacheChangeRequest> reqs,
- boolean failIfExists
+ Collection<DynamicCacheChangeRequest> reqs
) {
Collection<DynamicCacheStartFuture> res = new ArrayList<>(reqs.size());
Collection<DynamicCacheChangeRequest> sndReqs = new ArrayList<>(reqs.size());
for (DynamicCacheChangeRequest req : reqs) {
- DynamicCacheStartFuture fut = new DynamicCacheStartFuture(req.cacheName(), req);
+ DynamicCacheStartFuture fut = new DynamicCacheStartFuture(req.requestId());
try {
- if (req.stop() || req.close()) {
+ if (req.stop()) {
DynamicCacheDescriptor desc = cacheDescriptor(req.cacheName());
if (desc == null)
// No-op.
fut.onDone(false);
- else {
- assert desc.cacheConfiguration() != null : desc;
-
- if (req.close() && desc.cacheConfiguration().getCacheMode() == LOCAL) {
- req.close(false);
-
- req.stop(true);
- }
- }
}
+
if (req.start() && req.startCacheConfiguration() != null) {
CacheConfiguration ccfg = req.startCacheConfiguration();
@@ -2821,14 +2913,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
try {
ctx.discovery().sendCustomEvent(new DynamicCacheChangeBatch(sndReqs));
- if (ctx.isStopping()) {
- err = new IgniteCheckedException("Failed to execute dynamic cache change request, " +
- "node is stopping.");
- }
- else if (ctx.clientDisconnected()) {
- err = new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
- "Failed to execute dynamic cache change request, client node disconnected.");
- }
+ err = checkNodeState();
}
catch (IgniteCheckedException e) {
err = e;
@@ -2844,6 +2929,22 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
+ * @return Non null exception if node is stopping or disconnected.
+ */
+ @Nullable private IgniteCheckedException checkNodeState() {
+ if (ctx.isStopping()) {
+ return new IgniteCheckedException("Failed to execute dynamic cache change request, " +
+ "node is stopping.");
+ }
+ else if (ctx.clientDisconnected()) {
+ return new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
+ "Failed to execute dynamic cache change request, client node disconnected.");
+ }
+
+ return null;
+ }
+
+ /**
* @param type Event type.
* @param node Event node.
* @param topVer Topology version.
@@ -2859,9 +2960,10 @@ public class GridCacheProcessor extends GridProcessorAdapter {
*
* @param msg Customer message.
* @param topVer Current topology version.
+ * @param node Node sent message.
* @return {@code True} if minor topology version should be increased.
*/
- public boolean onCustomEvent(DiscoveryCustomMessage msg, AffinityTopologyVersion topVer) {
+ public boolean onCustomEvent(DiscoveryCustomMessage msg, AffinityTopologyVersion topVer, ClusterNode node) {
if (msg instanceof SchemaAbstractDiscoveryMessage) {
ctx.query().onDiscovery((SchemaAbstractDiscoveryMessage)msg);
@@ -2878,6 +2980,9 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (msg instanceof DynamicCacheChangeBatch)
return cachesInfo.onCacheChangeRequested((DynamicCacheChangeBatch)msg, topVer);
+ if (msg instanceof ClientCacheChangeDiscoveryMessage)
+ cachesInfo.onClientCacheChange((ClientCacheChangeDiscoveryMessage)msg, node);
+
return false;
}
@@ -3156,13 +3261,13 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (log.isDebugEnabled())
log.debug("Getting public cache for name: " + cacheName);
- IgniteCacheProxy<?, ?> cache = jCacheProxies.get(cacheName);
-
DynamicCacheDescriptor desc = cacheDescriptor(cacheName);
if (desc != null && !desc.cacheType().userCache())
throw new IllegalStateException("Failed to get cache because it is a system cache: " + cacheName);
+ IgniteCacheProxy<?, ?> cache = jCacheProxies.get(cacheName);
+
if (cache == null) {
dynamicStartCache(null, cacheName, null, false, failIfNotStarted, checkThreadTx).get();
@@ -3731,33 +3836,20 @@ public class GridCacheProcessor extends GridProcessorAdapter {
*/
@SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor")
private class DynamicCacheStartFuture extends GridFutureAdapter<Boolean> {
- /** Cache name. */
- private String cacheName;
-
- /** Change request. */
- @GridToStringInclude
- private DynamicCacheChangeRequest req;
-
- /**
- * @param cacheName Cache name.
- * @param req Cache start request.
- */
- private DynamicCacheStartFuture(String cacheName, DynamicCacheChangeRequest req) {
- this.cacheName = cacheName;
- this.req = req;
- }
+ /** */
+ private UUID id;
/**
- * @return Request.
+ * @param id Future ID.
*/
- public DynamicCacheChangeRequest request() {
- return req;
+ private DynamicCacheStartFuture(UUID id) {
+ this.id = id;
}
/** {@inheritDoc} */
@Override public boolean onDone(@Nullable Boolean res, @Nullable Throwable err) {
// Make sure to remove future before completion.
- pendingFuts.remove(req.requestId(), this);
+ pendingFuts.remove(id, this);
return super.onDone(res, err);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ClientCacheDhtTopologyFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ClientCacheDhtTopologyFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ClientCacheDhtTopologyFuture.java
new file mode 100644
index 0000000..3166159
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/ClientCacheDhtTopologyFuture.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht;
+
+import java.util.Collection;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ * Topology future created for client cache start.
+ */
+public class ClientCacheDhtTopologyFuture extends GridDhtTopologyFutureAdapter
+ implements GridDhtTopologyFuture {
+ /** */
+ final AffinityTopologyVersion topVer;
+
+ /**
+ * @param topVer Topology version.
+ */
+ public ClientCacheDhtTopologyFuture(AffinityTopologyVersion topVer) {
+ assert topVer != null;
+
+ this.topVer = topVer;
+
+ onDone(topVer);
+ }
+
+ /**
+ * @param topVer Topology version.
+ * @param e Error.
+ */
+ public ClientCacheDhtTopologyFuture(AffinityTopologyVersion topVer, IgniteCheckedException e) {
+ assert e != null;
+ assert topVer != null;
+
+ this.topVer = topVer;
+
+ onDone(e);
+ }
+
+ /**
+ * @param grp Cache group.
+ * @param topNodes Topology nodes.
+ */
+ public void validate(CacheGroupContext grp, Collection<ClusterNode> topNodes) {
+ grpValidRes = U.newHashMap(1);
+
+ grpValidRes.put(grp.groupId(), validateCacheGroup(grp,topNodes));
+ }
+
+ /** {@inheritDoc} */
+ @Override public AffinityTopologyVersion topologyVersion() {
+ return topVer;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "ClientCacheDhtTopologyFuture [topVer=" + topologyVersion() + ']';
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index 619630f..2d9e4f5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -83,7 +83,7 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
private Map<Integer, Set<UUID>> part2node = new HashMap<>();
/** */
- private GridDhtPartitionExchangeId lastExchangeId;
+ private AffinityTopologyVersion lastExchangeVer;
/** */
private AffinityTopologyVersion topVer = AffinityTopologyVersion.NONE;
@@ -183,21 +183,23 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public void updateTopologyVersion(
- GridDhtPartitionExchangeId exchId,
- GridDhtPartitionsExchangeFuture exchFut,
+ GridDhtTopologyFuture exchFut,
+ DiscoCache discoCache,
long updSeq,
boolean stopping
) throws IgniteInterruptedCheckedException {
U.writeLock(lock);
try {
- assert exchId.topologyVersion().compareTo(topVer) > 0 : "Invalid topology version [topVer=" + topVer +
- ", exchId=" + exchId + ']';
+ AffinityTopologyVersion exchTopVer = exchFut.topologyVersion();
+
+ assert exchTopVer.compareTo(topVer) > 0 : "Invalid topology version [topVer=" + topVer +
+ ", exchVer=" + exchTopVer + ']';
this.stopping = stopping;
- topVer = exchId.topologyVersion();
- discoCache = exchFut.discoCache();
+ topVer = exchTopVer;
+ this.discoCache = discoCache;
updateSeq.setIfGreater(updSeq);
@@ -560,19 +562,20 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@SuppressWarnings({"MismatchedQueryAndUpdateOfCollection"})
- @Nullable @Override public GridDhtPartitionMap update(@Nullable GridDhtPartitionExchangeId exchId,
+ @Nullable @Override public GridDhtPartitionMap update(
+ @Nullable AffinityTopologyVersion exchVer,
GridDhtPartitionFullMap partMap,
Map<Integer, T2<Long, Long>> cntrMap) {
if (log.isDebugEnabled())
- log.debug("Updating full partition map [exchId=" + exchId + ", parts=" + fullMapString() + ']');
+ log.debug("Updating full partition map [exchVer=" + exchVer + ", parts=" + fullMapString() + ']');
lock.writeLock().lock();
try {
- if (exchId != null && lastExchangeId != null && lastExchangeId.compareTo(exchId) >= 0) {
+ if (exchVer != null && lastExchangeVer != null && lastExchangeVer.compareTo(exchVer) >= 0) {
if (log.isDebugEnabled())
log.debug("Stale exchange id for full partition map update (will ignore) [lastExchId=" +
- lastExchangeId + ", exchId=" + exchId + ']');
+ lastExchangeVer + ", exchVer=" + exchVer + ']');
return null;
}
@@ -580,15 +583,15 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
if (node2part != null && node2part.compareTo(partMap) >= 0) {
if (log.isDebugEnabled())
log.debug("Stale partition map for full partition map update (will ignore) [lastExchId=" +
- lastExchangeId + ", exchId=" + exchId + ", curMap=" + node2part + ", newMap=" + partMap + ']');
+ lastExchangeVer + ", exchVer=" + exchVer + ", curMap=" + node2part + ", newMap=" + partMap + ']');
return null;
}
updateSeq.incrementAndGet();
- if (exchId != null)
- lastExchangeId = exchId;
+ if (exchVer != null)
+ lastExchangeVer = exchVer;
if (node2part != null) {
for (GridDhtPartitionMap part : node2part.values()) {
@@ -598,7 +601,7 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
// then we keep the newer value.
if (newPart != null && newPart.updateSequence() < part.updateSequence()) {
if (log.isDebugEnabled())
- log.debug("Overriding partition map in full update map [exchId=" + exchId + ", curPart=" +
+ log.debug("Overriding partition map in full update map [exchVer=" + exchVer + ", curPart=" +
mapString(part) + ", newPart=" + mapString(newPart) + ']');
partMap.put(part.nodeId(), part);
@@ -694,16 +697,16 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
if (stopping)
return null;
- if (lastExchangeId != null && exchId != null && lastExchangeId.compareTo(exchId) > 0) {
+ if (lastExchangeVer != null && exchId != null && lastExchangeVer.compareTo(exchId.topologyVersion()) > 0) {
if (log.isDebugEnabled())
- log.debug("Stale exchange id for single partition map update (will ignore) [lastExchId=" +
- lastExchangeId + ", exchId=" + exchId + ']');
+ log.debug("Stale exchange id for single partition map update (will ignore) [lastExchVer=" +
+ lastExchangeVer + ", exchId=" + exchId + ']');
return null;
}
if (exchId != null)
- lastExchangeId = exchId;
+ lastExchangeVer = exchId.topologyVersion();
if (node2part == null)
// Create invalid partition map.
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentRequest.java
index d9d642a..44c7b88 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentRequest.java
@@ -32,6 +32,12 @@ public class GridDhtAffinityAssignmentRequest extends GridCacheGroupIdMessage {
private static final long serialVersionUID = 0L;
/** */
+ private static final int SND_PART_STATE_MASK = 0x01;
+
+ /** */
+ private byte flags;
+
+ /** */
private long futId;
/** Topology version being queried. */
@@ -48,16 +54,28 @@ public class GridDhtAffinityAssignmentRequest extends GridCacheGroupIdMessage {
* @param futId Future ID.
* @param grpId Cache group ID.
* @param topVer Topology version.
+ * @param sndPartMap {@code True} if need send in response cache partitions state.
*/
public GridDhtAffinityAssignmentRequest(
long futId,
int grpId,
- AffinityTopologyVersion topVer) {
+ AffinityTopologyVersion topVer,
+ boolean sndPartMap) {
assert topVer != null;
this.futId = futId;
this.grpId = grpId;
this.topVer = topVer;
+
+ if (sndPartMap)
+ flags |= SND_PART_STATE_MASK;
+ }
+
+ /**
+ * @return {@code True} if need send in response cache partitions state.
+ */
+ public boolean sendPartitionsState() {
+ return (flags & SND_PART_STATE_MASK) != 0;
}
/**
@@ -91,7 +109,7 @@ public class GridDhtAffinityAssignmentRequest extends GridCacheGroupIdMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 5;
+ return 6;
}
/** {@inheritDoc} */
@@ -110,12 +128,18 @@ public class GridDhtAffinityAssignmentRequest extends GridCacheGroupIdMessage {
switch (writer.state()) {
case 3:
- if (!writer.writeLong("futId", futId))
+ if (!writer.writeByte("flags", flags))
return false;
writer.incrementState();
case 4:
+ if (!writer.writeLong("futId", futId))
+ return false;
+
+ writer.incrementState();
+
+ case 5:
if (!writer.writeMessage("topVer", topVer))
return false;
@@ -138,7 +162,7 @@ public class GridDhtAffinityAssignmentRequest extends GridCacheGroupIdMessage {
switch (reader.state()) {
case 3:
- futId = reader.readLong("futId");
+ flags = reader.readByte("flags");
if (!reader.isLastRead())
return false;
@@ -146,6 +170,14 @@ public class GridDhtAffinityAssignmentRequest extends GridCacheGroupIdMessage {
reader.incrementState();
case 4:
+ futId = reader.readLong("futId");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 5:
topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
index 4df3fc1..5b0de08 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAffinityAssignmentResponse.java
@@ -21,19 +21,20 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
-
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.GridDirectTransient;
-import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheGroupIdMessage;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.apache.ignite.plugin.extensions.communication.MessageWriter;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Affinity assignment response.
@@ -62,6 +63,13 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
/** Affinity assignment bytes. */
private byte[] idealAffAssignmentBytes;
+ /** */
+ @GridDirectTransient
+ private GridDhtPartitionFullMap partMap;
+
+ /** */
+ private byte[] partBytes;
+
/**
* Empty constructor.
*/
@@ -107,30 +115,30 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
}
/**
- * @param disco Discovery manager.
+ * @param discoCache Discovery data cache.
* @return Affinity assignment.
*/
- public List<List<ClusterNode>> affinityAssignment(GridDiscoveryManager disco) {
+ public List<List<ClusterNode>> affinityAssignment(DiscoCache discoCache) {
if (affAssignmentIds != null)
- return nodes(disco, affAssignmentIds);
+ return nodes(discoCache, affAssignmentIds);
return null;
}
/**
- * @param disco Discovery manager.
+ * @param discoCache Discovery data cache.
* @return Ideal affinity assignment.
*/
- public List<List<ClusterNode>> idealAffinityAssignment(GridDiscoveryManager disco) {
- return nodes(disco, idealAffAssignment);
+ public List<List<ClusterNode>> idealAffinityAssignment(DiscoCache discoCache) {
+ return nodes(discoCache, idealAffAssignment);
}
/**
- * @param disco Discovery manager.
+ * @param discoCache Discovery data cache.
* @param assignmentIds Assignment node IDs.
* @return Assignment nodes.
*/
- private List<List<ClusterNode>> nodes(GridDiscoveryManager disco, List<List<UUID>> assignmentIds) {
+ private List<List<ClusterNode>> nodes(DiscoCache discoCache, List<List<UUID>> assignmentIds) {
if (assignmentIds != null) {
List<List<ClusterNode>> assignment = new ArrayList<>(assignmentIds.size());
@@ -139,7 +147,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
List<ClusterNode> nodes = new ArrayList<>(ids.size());
for (int j = 0; j < ids.size(); j++) {
- ClusterNode node = disco.node(topVer, ids.get(j));
+ ClusterNode node = discoCache.node(ids.get(j));
assert node != null;
@@ -163,6 +171,20 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
}
/**
+ * @param partMap Partition map.
+ */
+ public void partitionMap(GridDhtPartitionFullMap partMap) {
+ this.partMap = partMap;
+ }
+
+ /**
+ * @return Partition map.
+ */
+ @Nullable public GridDhtPartitionFullMap partitionMap() {
+ return partMap;
+ }
+
+ /**
* @param assignments Assignment.
* @return Assignment where cluster nodes are converted to their ids.
*/
@@ -193,7 +215,7 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 7;
+ return 8;
}
/**
@@ -208,6 +230,9 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
if (idealAffAssignment != null && idealAffAssignmentBytes == null)
idealAffAssignmentBytes = U.marshal(ctx, idealAffAssignment);
+
+ if (partMap != null && partBytes == null)
+ partBytes = U.zip(U.marshal(ctx.marshaller(), partMap));
}
/** {@inheritDoc} */
@@ -222,6 +247,9 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
if (idealAffAssignmentBytes != null && idealAffAssignment == null)
idealAffAssignment = U.unmarshal(ctx, idealAffAssignmentBytes, ldr);
+
+ if (partBytes != null && partMap == null)
+ partMap = U.unmarshalZip(ctx.marshaller(), partBytes, U.resolveClassLoader(ldr, ctx.gridConfig()));
}
/** {@inheritDoc} */
@@ -263,6 +291,12 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
writer.incrementState();
case 6:
+ if (!writer.writeByteArray("partBytes", partBytes))
+ return false;
+
+ writer.incrementState();
+
+ case 7:
if (!writer.writeMessage("topVer", topVer))
return false;
@@ -309,6 +343,14 @@ public class GridDhtAffinityAssignmentResponse extends GridCacheGroupIdMessage {
reader.incrementState();
case 6:
+ partBytes = reader.readByteArray("partBytes");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 7:
topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
index 8746320..dcc08d4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtAssignmentFetchFuture.java
@@ -27,13 +27,12 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.internal.IgniteNeedReconnectException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.GridNodeOrderComparator;
+import org.apache.ignite.internal.IgniteNeedReconnectException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
-import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -76,25 +75,28 @@ public class GridDhtAssignmentFetchFuture extends GridFutureAdapter<GridDhtAffin
/** */
private final int grpId;
+ /** */
+ private boolean needPartState;
+
/**
* @param ctx Context.
- * @param grpDesc Group descriptor.
+ * @param grpId Group ID.
* @param topVer Topology version.
* @param discoCache Discovery cache.
*/
public GridDhtAssignmentFetchFuture(
GridCacheSharedContext ctx,
- CacheGroupDescriptor grpDesc,
+ int grpId,
AffinityTopologyVersion topVer,
DiscoCache discoCache
) {
this.topVer = topVer;
- this.grpId = grpDesc.groupId();
+ this.grpId = grpId;
this.ctx = ctx;
id = idGen.getAndIncrement();
- Collection<ClusterNode> availableNodes = discoCache.cacheGroupAffinityNodes(grpDesc.groupId());
+ Collection<ClusterNode> availableNodes = discoCache.cacheGroupAffinityNodes(grpId);
LinkedList<ClusterNode> tmp = new LinkedList<>();
@@ -127,8 +129,12 @@ public class GridDhtAssignmentFetchFuture extends GridFutureAdapter<GridDhtAffin
/**
* Initializes fetch future.
+ *
+ * @param needPartState {@code True} if also need fetch partitions state.
*/
- public void init() {
+ public void init(boolean needPartState) {
+ this.needPartState = needPartState;
+
ctx.affinity().addDhtAssignmentFetchFuture(this);
requestFromNextNode();
@@ -195,7 +201,7 @@ public class GridDhtAssignmentFetchFuture extends GridFutureAdapter<GridDhtAffin
", node=" + node + ']');
ctx.io().send(node,
- new GridDhtAffinityAssignmentRequest(id, grpId, topVer),
+ new GridDhtAffinityAssignmentRequest(id, grpId, topVer, needPartState),
AFFINITY_POOL);
// Close window for listener notification.
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
index 05831b1..ed80f83 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheEntry.java
@@ -407,14 +407,6 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
return null;
}
- // If remote node has no near cache, don't add it.
- if (!cctx.discovery().cacheNearNode(node, cacheName())) {
- if (log.isDebugEnabled())
- log.debug("Ignoring near reader because near cache is disabled: " + nodeId);
-
- return null;
- }
-
// If remote node is (primary?) or back up, don't add it as a reader.
if (cctx.affinity().partitionBelongs(node, partition(), topVer)) {
if (log.isDebugEnabled())
@@ -653,7 +645,7 @@ public class GridDhtCacheEntry extends GridDistributedCacheEntry {
for (int i = 0; i < rdrs.length; i++) {
ClusterNode node = cctx.discovery().getAlive(rdrs[i].nodeId());
- if (node == null || !cctx.discovery().cacheNode(node, cacheName())) {
+ if (node == null) {
// Node has left and if new list has already been created, just skip.
// Otherwise, create new list and add alive nodes.
if (newRdrs == null) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
index d365a8e..acb822c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java
@@ -26,6 +26,7 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
@@ -54,15 +55,15 @@ public interface GridDhtPartitionTopology {
/**
* Updates topology version.
*
- * @param exchId Exchange ID.
* @param exchFut Exchange future.
+ * @param discoCache Discovery data cache.
* @param updateSeq Update sequence.
* @param stopping Stopping flag.
* @throws IgniteInterruptedCheckedException If interrupted.
*/
public void updateTopologyVersion(
- GridDhtPartitionExchangeId exchId,
- GridDhtPartitionsExchangeFuture exchFut,
+ GridDhtTopologyFuture exchFut,
+ DiscoCache discoCache,
long updateSeq,
boolean stopping
) throws IgniteInterruptedCheckedException;
@@ -221,12 +222,13 @@ public interface GridDhtPartitionTopology {
public void onRemoved(GridDhtCacheEntry e);
/**
- * @param exchId Exchange ID.
+ * @param exchangeVer Exchange version.
* @param partMap Update partition map.
* @param cntrMap Partition update counters.
* @return Local partition map if there were evictions or {@code null} otherwise.
*/
- public GridDhtPartitionMap update(@Nullable GridDhtPartitionExchangeId exchId,
+ public GridDhtPartitionMap update(
+ @Nullable AffinityTopologyVersion exchangeVer,
GridDhtPartitionFullMap partMap,
@Nullable Map<Integer, T2<Long, Long>> cntrMap);
@@ -250,7 +252,7 @@ public interface GridDhtPartitionTopology {
* This method should be called on topology coordinator after all partition messages are received.
*
* @param discoEvt Discovery event for which we detect lost partitions.
- * @return {@code True} if partitons state got updated.
+ * @return {@code True} if partitions state got updated.
*/
public boolean detectLostPartitions(DiscoveryEvent discoEvt);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index 89554f3..ea731f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -105,7 +105,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
private Map<Integer, Set<UUID>> part2node = new HashMap<>();
/** */
- private GridDhtPartitionExchangeId lastExchangeId;
+ private AffinityTopologyVersion lastExchangeVer;
/** */
private volatile AffinityTopologyVersion topVer = AffinityTopologyVersion.NONE;
@@ -167,7 +167,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
part2node = new HashMap<>();
- lastExchangeId = null;
+ lastExchangeVer = null;
updateSeq.set(1);
@@ -301,16 +301,18 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@Override public void updateTopologyVersion(
- GridDhtPartitionExchangeId exchId,
- GridDhtPartitionsExchangeFuture exchFut,
+ GridDhtTopologyFuture exchFut,
+ DiscoCache discoCache,
long updSeq,
boolean stopping
) throws IgniteInterruptedCheckedException {
U.writeLock(lock);
try {
- assert exchId.topologyVersion().compareTo(topVer) > 0 : "Invalid topology version [topVer=" + topVer +
- ", exchId=" + exchId +
+ AffinityTopologyVersion exchTopVer = exchFut.topologyVersion();
+
+ assert exchTopVer.compareTo(topVer) > 0 : "Invalid topology version [topVer=" + topVer +
+ ", exchTopVer=" + exchTopVer +
", fut=" + exchFut + ']';
this.stopping = stopping;
@@ -321,9 +323,9 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
rebalancedTopVer = AffinityTopologyVersion.NONE;
- topVer = exchId.topologyVersion();
+ topVer = exchTopVer;
- discoCache = exchFut.discoCache();
+ this.discoCache = discoCache;
}
finally {
lock.writeLock().unlock();
@@ -1109,12 +1111,12 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
/** {@inheritDoc} */
@SuppressWarnings({"MismatchedQueryAndUpdateOfCollection"})
@Override public GridDhtPartitionMap update(
- @Nullable GridDhtPartitionExchangeId exchId,
+ @Nullable AffinityTopologyVersion exchangeVer,
GridDhtPartitionFullMap partMap,
@Nullable Map<Integer, T2<Long, Long>> cntrMap
) {
if (log.isDebugEnabled())
- log.debug("Updating full partition map [exchId=" + exchId + ", parts=" + fullMapString() + ']');
+ log.debug("Updating full partition map [exchVer=" + exchangeVer + ", parts=" + fullMapString() + ']');
assert partMap != null;
@@ -1147,27 +1149,26 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
}
}
- //if need skip
- if (exchId != null && lastExchangeId != null && lastExchangeId.compareTo(exchId) >= 0) {
+ if (exchangeVer != null && lastExchangeVer != null && lastExchangeVer.compareTo(exchangeVer) >= 0) {
if (log.isDebugEnabled())
- log.debug("Stale exchange id for full partition map update (will ignore) [lastExchId=" +
- lastExchangeId + ", exchId=" + exchId + ']');
+ log.debug("Stale exchange id for full partition map update (will ignore) [lastExch=" +
+ lastExchangeVer + ", exch=" + exchangeVer + ']');
return null;
}
if (node2part != null && node2part.compareTo(partMap) >= 0) {
if (log.isDebugEnabled())
- log.debug("Stale partition map for full partition map update (will ignore) [lastExchId=" +
- lastExchangeId + ", exchId=" + exchId + ", curMap=" + node2part + ", newMap=" + partMap + ']');
+ log.debug("Stale partition map for full partition map update (will ignore) [lastExch=" +
+ lastExchangeVer + ", exch=" + exchangeVer + ", curMap=" + node2part + ", newMap=" + partMap + ']');
return null;
}
long updateSeq = this.updateSeq.incrementAndGet();
- if (exchId != null)
- lastExchangeId = exchId;
+ if (exchangeVer != null)
+ lastExchangeVer = exchangeVer;
if (node2part != null) {
for (GridDhtPartitionMap part : node2part.values()) {
@@ -1180,8 +1181,8 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
(grp.localStartVersion().compareTo(newPart.topologyVersion()) > 0))
) {
if (log.isDebugEnabled())
- log.debug("Overriding partition map in full update map [exchId=" + exchId + ", curPart=" +
- mapString(part) + ", newPart=" + mapString(newPart) + ']');
+ log.debug("Overriding partition map in full update map [exch=" + exchangeVer +
+ ", curPart=" + mapString(part) + ", newPart=" + mapString(newPart) + ']');
partMap.put(part.nodeId(), part);
}
@@ -1333,16 +1334,16 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
if (stopping)
return null;
- if (lastExchangeId != null && exchId != null && lastExchangeId.compareTo(exchId) > 0) {
+ if (lastExchangeVer != null && exchId != null && lastExchangeVer.compareTo(exchId.topologyVersion()) > 0) {
if (log.isDebugEnabled())
- log.debug("Stale exchange id for single partition map update (will ignore) [lastExchId=" +
- lastExchangeId + ", exchId=" + exchId + ']');
+ log.debug("Stale exchange id for single partition map update (will ignore) [lastExch=" +
+ lastExchangeVer + ", exch=" + exchId.topologyVersion() + ']');
return null;
}
if (exchId != null)
- lastExchangeId = exchId;
+ lastExchangeVer = exchId.topologyVersion();
if (node2part == null)
// Create invalid partition map.
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFutureAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFutureAdapter.java
new file mode 100644
index 0000000..e70f383
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTopologyFutureAdapter.java
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed.dht;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.PartitionLossPolicy;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.TopologyValidator;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.CacheInvalidStateException;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_ALL;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_SAFE;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_ALL;
+import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE;
+
+/**
+ *
+ */
+public abstract class GridDhtTopologyFutureAdapter extends GridFutureAdapter<AffinityTopologyVersion>
+ implements GridDhtTopologyFuture {
+ /** Cache groups validation results. */
+ protected volatile Map<Integer, CacheValidation> grpValidRes;
+
+ /**
+ * @param grp Cache group.
+ * @param topNodes Topology nodes.
+ * @return Validation result.
+ */
+ protected final CacheValidation validateCacheGroup(CacheGroupContext grp, Collection<ClusterNode> topNodes) {
+ Collection<Integer> lostParts = grp.isLocal() ?
+ Collections.<Integer>emptyList() : grp.topology().lostPartitions();
+
+ boolean valid = true;
+
+ if (!grp.systemCache()) {
+ TopologyValidator validator = grp.topologyValidator();
+
+ if (validator != null)
+ valid = validator.validate(topNodes);
+ }
+
+ return new CacheValidation(valid, lostParts);
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public final Throwable validateCache(
+ GridCacheContext cctx,
+ boolean recovery,
+ boolean read,
+ @Nullable Object key,
+ @Nullable Collection<?> keys
+ ) {
+ assert isDone() : this;
+
+ Throwable err = error();
+
+ if (err != null)
+ return err;
+
+ if (!cctx.shared().kernalContext().state().active())
+ return new CacheInvalidStateException(
+ "Failed to perform cache operation (cluster is not activated): " + cctx.name());
+
+ CacheGroupContext grp = cctx.group();
+
+ PartitionLossPolicy partLossPlc = grp.config().getPartitionLossPolicy();
+
+ if (grp.needsRecovery() && !recovery) {
+ if (!read && (partLossPlc == READ_ONLY_SAFE || partLossPlc == READ_ONLY_ALL))
+ return new IgniteCheckedException("Failed to write to cache (cache is moved to a read-only state): " +
+ cctx.name());
+ }
+
+ if (grp.needsRecovery() || grp.topologyValidator() != null) {
+ CacheValidation validation = grpValidRes.get(grp.groupId());
+
+ if (validation == null)
+ return null;
+
+ if (!validation.valid && !read)
+ return new IgniteCheckedException("Failed to perform cache operation " +
+ "(cache topology is not valid): " + cctx.name());
+
+ if (recovery || !grp.needsRecovery())
+ return null;
+
+ if (key != null) {
+ int p = cctx.affinity().partition(key);
+
+ CacheInvalidStateException ex = validatePartitionOperation(cctx.name(), read, key, p,
+ validation.lostParts, partLossPlc);
+
+ if (ex != null)
+ return ex;
+ }
+
+ if (keys != null) {
+ for (Object k : keys) {
+ int p = cctx.affinity().partition(k);
+
+ CacheInvalidStateException ex = validatePartitionOperation(cctx.name(), read, k, p,
+ validation.lostParts, partLossPlc);
+
+ if (ex != null)
+ return ex;
+ }
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * @param cacheName Cache name.
+ * @param read Read flag.
+ * @param key Key to check.
+ * @param part Partition this key belongs to.
+ * @param lostParts Collection of lost partitions.
+ * @param plc Partition loss policy.
+ * @return Invalid state exception if this operation is disallowed.
+ */
+ private CacheInvalidStateException validatePartitionOperation(
+ String cacheName,
+ boolean read,
+ Object key,
+ int part,
+ Collection<Integer> lostParts,
+ PartitionLossPolicy plc
+ ) {
+ if (lostParts.contains(part)) {
+ if (!read) {
+ assert plc == READ_WRITE_ALL || plc == READ_WRITE_SAFE;
+
+ if (plc == READ_WRITE_SAFE) {
+ return new CacheInvalidStateException("Failed to execute cache operation " +
+ "(all partition owners have left the grid, partition data has been lost) [" +
+ "cacheName=" + cacheName + ", part=" + part + ", key=" + key + ']');
+ }
+ }
+ else {
+ // Read.
+ if (plc == READ_ONLY_SAFE || plc == READ_WRITE_SAFE)
+ return new CacheInvalidStateException("Failed to execute cache operation " +
+ "(all partition owners have left the grid, partition data has been lost) [" +
+ "cacheName=" + cacheName + ", part=" + part + ", key=" + key + ']');
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * Cache validation result.
+ */
+ protected static class CacheValidation {
+ /** Topology validation result. */
+ private boolean valid;
+
+ /** Lost partitions on this topology version. */
+ private Collection<Integer> lostParts;
+
+ /**
+ * @param valid Valid flag.
+ * @param lostParts Lost partitions.
+ */
+ private CacheValidation(boolean valid, Collection<Integer> lostParts) {
+ this.valid = valid;
+ this.lostParts = lostParts;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
index c7c4280..631fe10 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
@@ -19,10 +19,12 @@ package org.apache.ignite.internal.processors.cache.distributed.dht;
import java.io.Externalizable;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
@@ -97,6 +99,10 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** Preload keys. */
private BitSet preloadKeys;
+ /** */
+ @GridDirectTransient
+ private List<IgniteTxKey> nearWritesCacheMissed;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -163,6 +169,13 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
}
/**
+ * @return Near cache writes for which cache was not found (possible if client near cache was closed).
+ */
+ @Nullable public List<IgniteTxKey> nearWritesCacheMissed() {
+ return nearWritesCacheMissed;
+ }
+
+ /**
* @return Near transaction ID.
*/
public GridCacheVersion nearXidVersion() {
@@ -319,13 +332,37 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
while (keyIter.hasNext()) {
IgniteTxKey key = keyIter.next();
- key.finishUnmarshal(ctx.cacheContext(key.cacheId()), ldr);
+ GridCacheContext<?, ?> cacheCtx = ctx.cacheContext(key.cacheId());
+
+ if (cacheCtx != null) {
+ key.finishUnmarshal(cacheCtx, ldr);
- owned.put(key, valIter.next());
+ owned.put(key, valIter.next());
+ }
}
}
- unmarshalTx(nearWrites, true, ctx, ldr);
+ if (nearWrites != null) {
+ for (Iterator<IgniteTxEntry> it = nearWrites.iterator(); it.hasNext();) {
+ IgniteTxEntry e = it.next();
+
+ GridCacheContext<?, ?> cacheCtx = ctx.cacheContext(e.cacheId());
+
+ if (cacheCtx == null) {
+ it.remove();
+
+ if (nearWritesCacheMissed == null)
+ nearWritesCacheMissed = new ArrayList<>();
+
+ nearWritesCacheMissed.add(e.txKey());
+ }
+ else {
+ e.context(cacheCtx);
+
+ e.unmarshal(ctx, true, ldr);
+ }
+ }
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
index 5dcb98c..0c2bf81 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareResponse.java
@@ -124,7 +124,7 @@ public class GridDhtTxPrepareResponse extends GridDistributedTxPrepareResponse {
/**
* @return Evicted readers.
*/
- Collection<IgniteTxKey> nearEvicted() {
+ public Collection<IgniteTxKey> nearEvicted() {
return nearEvicted;
}
@@ -194,7 +194,9 @@ public class GridDhtTxPrepareResponse extends GridDistributedTxPrepareResponse {
for (IgniteTxKey key : nearEvicted) {
GridCacheContext cctx = ctx.cacheContext(key.cacheId());
- key.prepareMarshal(cctx);
+ // Can be null if client near cache was removed, in this case assume do not need prepareMarshal.
+ if (cctx != null)
+ key.prepareMarshal(cctx);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 67e3ebc..52f007a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -3240,9 +3240,17 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
GridDhtAtomicUpdateResponse dhtRes = null;
- if (isNearEnabled(cacheCfg)) {
- List<KeyCacheObject> nearEvicted =
- ((GridNearAtomicCache<K, V>)near()).processDhtAtomicUpdateRequest(nodeId, req, nearRes);
+ if (req.nearSize() > 0) {
+ List<KeyCacheObject> nearEvicted;
+
+ if (isNearEnabled(ctx))
+ nearEvicted = ((GridNearAtomicCache<K, V>)near()).processDhtAtomicUpdateRequest(nodeId, req, nearRes);
+ else {
+ nearEvicted = new ArrayList<>(req.nearSize());
+
+ for (int i = 0; i < req.nearSize(); i++)
+ nearEvicted.add(req.nearKey(i));
+ }
if (nearEvicted != null) {
dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
index 7b2547a..70bf6f5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateResponse.java
@@ -118,7 +118,7 @@ public class GridDhtAtomicUpdateResponse extends GridCacheIdMessage implements G
/**
* @param nearEvicted Evicted near cache keys.
*/
- void nearEvicted(List<KeyCacheObject> nearEvicted) {
+ public void nearEvicted(List<KeyCacheObject> nearEvicted) {
this.nearEvicted = nearEvicted;
}
@@ -133,10 +133,13 @@ public class GridDhtAtomicUpdateResponse extends GridCacheIdMessage implements G
GridCacheContext cctx = ctx.cacheContext(cacheId);
- prepareMarshalCacheObjects(nearEvicted, cctx);
+ // Can be null if client near cache was removed, in this case assume do not need prepareMarshal.
+ if (cctx != null) {
+ prepareMarshalCacheObjects(nearEvicted, cctx);
- if (errs != null)
- errs.prepareMarshal(this, cctx);
+ if (errs != null)
+ errs.prepareMarshal(this, cctx);
+ }
}
/** {@inheritDoc} */
[4/4] ignite git commit: ignite-5272 Do not use synchronous custom
discovery event for client cache start/close
Posted by sb...@apache.org.
ignite-5272 Do not use synchronous custom discovery event for client cache start/close
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/4026ddcc
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/4026ddcc
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/4026ddcc
Branch: refs/heads/master
Commit: 4026ddcc2161ca4bf5ad3dc04cabd284c608738a
Parents: 3fd9e04
Author: sboikov <sb...@gridgain.com>
Authored: Thu Jun 15 09:23:36 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Jun 15 09:23:36 2017 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 12 +
.../internal/managers/discovery/DiscoCache.java | 17 -
.../discovery/GridDiscoveryManager.java | 88 ++-
.../affinity/GridAffinityProcessor.java | 24 +-
.../cache/CacheAffinitySharedManager.java | 612 +++++++++++++++----
.../processors/cache/CacheGroupContext.java | 3 +
.../ClientCacheChangeDiscoveryMessage.java | 176 ++++++
.../ClientCacheChangeDummyDiscoveryMessage.java | 104 ++++
.../cache/ClientCacheUpdateTimeout.java | 44 ++
.../processors/cache/ClusterCachesInfo.java | 94 ++-
.../cache/DynamicCacheChangeRequest.java | 30 -
.../processors/cache/ExchangeActions.java | 87 +--
.../processors/cache/GridCacheContext.java | 9 +-
.../processors/cache/GridCacheIoManager.java | 213 +++++--
.../GridCachePartitionExchangeManager.java | 11 +-
.../processors/cache/GridCacheProcessor.java | 364 ++++++-----
.../dht/ClientCacheDhtTopologyFuture.java | 78 +++
.../dht/GridClientPartitionTopology.java | 41 +-
.../dht/GridDhtAffinityAssignmentRequest.java | 40 +-
.../dht/GridDhtAffinityAssignmentResponse.java | 66 +-
.../dht/GridDhtAssignmentFetchFuture.java | 22 +-
.../distributed/dht/GridDhtCacheEntry.java | 10 +-
.../dht/GridDhtPartitionTopology.java | 14 +-
.../dht/GridDhtPartitionTopologyImpl.java | 49 +-
.../dht/GridDhtTopologyFutureAdapter.java | 195 ++++++
.../dht/GridDhtTxPrepareRequest.java | 43 +-
.../dht/GridDhtTxPrepareResponse.java | 6 +-
.../dht/atomic/GridDhtAtomicCache.java | 14 +-
.../dht/atomic/GridDhtAtomicUpdateResponse.java | 11 +-
.../GridDhtPartitionsExchangeFuture.java | 201 ++----
.../cache/transactions/IgniteTxEntry.java | 7 +
.../cache/transactions/IgniteTxHandler.java | 13 +-
.../cache/transactions/IgniteTxManager.java | 27 +
.../GridProjectionForCachesSelfTest.java | 24 +-
.../IgniteClientReconnectCacheTest.java | 29 +-
.../cache/CacheStopAndDestroySelfTest.java | 38 +-
.../cache/GridCacheClearSelfTest.java | 15 +-
...IgniteClientCacheInitializationFailTest.java | 6 +-
.../IgniteClientCacheStartFailoverTest.java | 585 ++++++++++++++++++
.../IgniteDynamicClientCacheStartSelfTest.java | 248 +++++++-
.../cache/IgniteNearClientCacheCloseTest.java | 262 ++++++++
...gniteTopologyValidatorAbstractCacheTest.java | 178 ++++--
...ologyValidatorAbstractTxCacheGroupsTest.java | 12 +-
...iteTopologyValidatorAbstractTxCacheTest.java | 28 +-
.../CacheLateAffinityAssignmentTest.java | 16 +-
...teCacheClientNodePartitionsExchangeTest.java | 23 +-
.../testframework/junits/GridAbstractTest.java | 2 +
.../testsuites/IgniteCacheTestSuite2.java | 4 +
48 files changed, 3270 insertions(+), 925 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index 539f288..d9c112c 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -23,6 +23,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import javax.net.ssl.HostnameVerifier;
+import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshaller;
import org.apache.ignite.lang.IgnitePredicate;
import org.jetbrains.annotations.Nullable;
@@ -623,6 +624,17 @@ public final class IgniteSystemProperties {
public static final String IGNITE_SECURITY_COMPATIBILITY_MODE = "IGNITE_SECURITY_COMPATIBILITY_MODE";
/**
+ * When client cache is started or closed special discovery message is sent to notify cluster (for example this is
+ * needed for {@link ClusterGroup#forCacheNodes(String)} API. This timeout specifies how long to wait
+ * after client cache start/close before sending this message. If during this timeout another client
+ * cache changed, these events are combined into single message.
+ * <p>
+ * Default is 10 seconds.
+ */
+ public static final String IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT =
+ "IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT";
+
+ /**
* Enforces singleton.
*/
private IgniteSystemProperties() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
index 22c2d07..2b3c4fc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
@@ -74,10 +74,6 @@ public class DiscoCache {
/** Node map. */
private final Map<UUID, ClusterNode> nodeMap;
- /** Caches where at least one node has near cache enabled. */
- @GridToStringInclude
- private final Set<Integer> nearEnabledCaches;
-
/** Alive nodes. */
private final Set<UUID> alives = new GridConcurrentHashSet<>();
@@ -93,7 +89,6 @@ public class DiscoCache {
* @param allCacheNodes Cache nodes by cache name.
* @param cacheGrpAffNodes Affinity nodes by cache group ID.
* @param nodeMap Node map.
- * @param nearEnabledCaches Caches where at least one node has near cache enabled.
* @param alives Alive nodes.
*/
DiscoCache(ClusterNode loc,
@@ -107,7 +102,6 @@ public class DiscoCache {
Map<Integer, List<ClusterNode>> allCacheNodes,
Map<Integer, List<ClusterNode>> cacheGrpAffNodes,
Map<UUID, ClusterNode> nodeMap,
- Set<Integer> nearEnabledCaches,
Set<UUID> alives) {
this.loc = loc;
this.rmtNodes = rmtNodes;
@@ -120,7 +114,6 @@ public class DiscoCache {
this.allCacheNodes = allCacheNodes;
this.cacheGrpAffNodes = cacheGrpAffNodes;
this.nodeMap = nodeMap;
- this.nearEnabledCaches = nearEnabledCaches;
this.alives.addAll(alives);
}
@@ -243,16 +236,6 @@ public class DiscoCache {
}
/**
- * Checks if cache with given ID has at least one node with near cache enabled.
- *
- * @param cacheId Cache ID.
- * @return {@code True} if cache with given name has at least one node with near cache enabled.
- */
- public boolean hasNearCache(int cacheId) {
- return nearEnabledCaches.contains(cacheId);
- }
-
- /**
* @param id Node ID.
* @return Node.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index c91ff74..0f0d1c9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -71,8 +71,11 @@ import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
+import org.apache.ignite.internal.processors.cache.ClientCacheChangeDummyDiscoveryMessage;
+import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest;
import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
@@ -357,11 +360,13 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/**
* Called from discovery thread. Adds dynamic cache filter.
*
+ * @param cacheId Cache ID.
* @param grpId Cache group ID.
* @param cacheName Cache name.
* @param nearEnabled Near enabled flag.
*/
public void setCacheFilter(
+ int cacheId,
int grpId,
String cacheName,
boolean nearEnabled
@@ -374,7 +379,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
if (grp.cacheMode == CacheMode.REPLICATED)
nearEnabled = false;
- registeredCaches.put(cacheName, new CachePredicate(grp, nearEnabled));
+ registeredCaches.put(cacheName, new CachePredicate(cacheId, grp, nearEnabled));
}
}
@@ -586,8 +591,9 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
assert customMsg != null;
boolean incMinorTopVer = ctx.cache().onCustomEvent(
- customMsg, new AffinityTopologyVersion(topVer, minorTopVer)
- );
+ customMsg,
+ new AffinityTopologyVersion(topVer, minorTopVer),
+ node);
if (incMinorTopVer) {
minorTopVer++;
@@ -1860,17 +1866,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
- * Checks if cache with given ID has at least one node with near cache enabled.
- *
- * @param cacheId Cache ID.
- * @param topVer Topology version.
- * @return {@code True} if cache with given name has at least one node with near cache enabled.
- */
- public boolean hasNearCache(int cacheId, AffinityTopologyVersion topVer) {
- return resolveDiscoCache(cacheId, topVer).hasNearCache(cacheId);
- }
-
- /**
* Gets discovery cache for given topology version.
*
* @param grpId Cache group ID (participates in exception message).
@@ -1994,6 +1989,22 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
}
/**
+ * @param reqId Start request ID.
+ * @param startReqs Cache start requests.
+ * @param cachesToClose Cache to close.
+ */
+ public void clientCacheStartEvent(UUID reqId,
+ @Nullable Map<String, DynamicCacheChangeRequest> startReqs,
+ @Nullable Set<String> cachesToClose) {
+ discoWrk.addEvent(DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT,
+ AffinityTopologyVersion.NONE,
+ localNode(),
+ null,
+ Collections.<ClusterNode>emptyList(),
+ new ClientCacheChangeDummyDiscoveryMessage(reqId, startReqs, cachesToClose));
+ }
+
+ /**
* Gets first grid node start time, see {@link DiscoverySpi#getGridStartTime()}.
*
* @return Start time of the first grid node.
@@ -2109,8 +2120,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
Set<ClusterNode> rmtNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
Set<ClusterNode> srvNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
- Set<Integer> nearEnabledCaches = new HashSet<>();
-
for (ClusterNode node : allNodes) {
assert node.order() != 0 : "Invalid node order [locNode=" + loc + ", node=" + node + ']';
assert !node.isDaemon();
@@ -2143,9 +2152,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
rmtNodesWithCaches.add(node);
addToMap(allCacheNodes, cacheName, node);
-
- if (filter.nearNode(node))
- nearEnabledCaches.add(CU.cacheId(cacheName));
}
}
}
@@ -2162,7 +2168,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
Collections.unmodifiableMap(allCacheNodes),
Collections.unmodifiableMap(cacheGrpAffNodes),
Collections.unmodifiableMap(nodeMap),
- Collections.unmodifiableSet(nearEnabledCaches),
alives);
}
@@ -2808,23 +2813,32 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/**
* Cache predicate.
*/
- private static class CachePredicate {
+ private class CachePredicate {
+ /** */
+ private final int cacheId;
+
/** Cache filter. */
private final CacheGroupAffinity aff;
/** If near cache is enabled on data nodes. */
private final boolean nearEnabled;
- /** Collection of client near nodes. */
+ /**
+ * Collection of client nodes.
+ *
+ * Note: if client cache started/closed this map is updated asynchronously.
+ */
private final ConcurrentHashMap<UUID, Boolean> clientNodes;
/**
+ * @param cacheId Cache ID.
* @param aff Cache group affinity.
* @param nearEnabled Near enabled flag.
*/
- private CachePredicate(CacheGroupAffinity aff, boolean nearEnabled) {
+ private CachePredicate(int cacheId, CacheGroupAffinity aff, boolean nearEnabled) {
assert aff != null;
+ this.cacheId = cacheId;
this.aff = aff;
this.nearEnabled = nearEnabled;
@@ -2836,7 +2850,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @param nearEnabled Near enabled flag.
* @return {@code True} if new node ID was added.
*/
- public boolean addClientNode(UUID nodeId, boolean nearEnabled) {
+ boolean addClientNode(UUID nodeId, boolean nearEnabled) {
assert nodeId != null;
Boolean old = clientNodes.putIfAbsent(nodeId, nearEnabled);
@@ -2868,19 +2882,20 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
* @param node Node to check.
* @return {@code True} if cache is accessible on the given node.
*/
- public boolean cacheNode(ClusterNode node) {
- return !node.isDaemon() && (CU.affinityNode(node, aff.cacheFilter) || clientNodes.containsKey(node.id()));
+ boolean cacheNode(ClusterNode node) {
+ return !node.isDaemon() && (CU.affinityNode(node, aff.cacheFilter) ||
+ cacheClientNode(node) != null);
}
/**
* @param node Node to check.
* @return {@code True} if near cache is present on the given nodes.
*/
- public boolean nearNode(ClusterNode node) {
+ boolean nearNode(ClusterNode node) {
if (CU.affinityNode(node, aff.cacheFilter))
return nearEnabled;
- Boolean near = clientNodes.get(node.id());
+ Boolean near = cacheClientNode(node);
return near != null && near;
}
@@ -2893,9 +2908,24 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
if (node.isDaemon())
return false;
- Boolean near = clientNodes.get(node.id());
+ Boolean near = cacheClientNode(node);
return near != null && !near;
}
+
+ /**
+ * @param node Node.
+ * @return {@code Null} if client cache does not exist, otherwise cache near enabled flag.
+ */
+ private Boolean cacheClientNode(ClusterNode node) {
+ // On local node check actual cache state since clientNodes map is updated asynchronously.
+ if (ctx.localNodeId().equals(node.id())) {
+ GridCacheContext cctx = ctx.cache().context().cacheContext(cacheId);
+
+ return cctx != null ? CU.isNearEnabled(cctx) : null;
+ }
+
+ return clientNodes.get(node.id());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
index 87c424a..6b43fc7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityProcessor.java
@@ -39,6 +39,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
@@ -364,18 +365,9 @@ public class GridAffinityProcessor extends GridProcessorAdapter {
if (fut != null)
return fut.get();
- ClusterNode loc = ctx.discovery().localNode();
-
- // Check local node.
- Collection<ClusterNode> cacheNodes = ctx.discovery().cacheNodes(cacheName, topVer);
-
- if (cacheNodes.contains(loc)) {
- GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(cacheName);
-
- // Cache is being stopped.
- if (cache == null)
- return null;
+ GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(cacheName);
+ if (cache != null) {
GridCacheContext<Object, Object> cctx = cache.context();
cctx.awaitStarted();
@@ -412,6 +404,8 @@ public class GridAffinityProcessor extends GridProcessorAdapter {
}
}
+ Collection<ClusterNode> cacheNodes = ctx.discovery().cacheNodes(cacheName, topVer);
+
if (F.isEmpty(cacheNodes))
return null;
@@ -443,7 +437,13 @@ public class GridAffinityProcessor extends GridProcessorAdapter {
CacheMode mode = ctx.cache().cacheMode(cacheName);
- assert mode != null;
+ if (mode == null) {
+ if (ctx.clientDisconnected())
+ throw new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
+ "Failed to get affinity mapping, client disconnected.");
+
+ throw new IgniteCheckedException("No cache nodes in topology for cache name: " + cacheName);
+ }
// Map all keys to a single node, if the cache mode is LOCAL.
if (mode == LOCAL) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index b6fe90b..0f62db2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -28,7 +29,9 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import javax.cache.CacheException;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.affinity.AffinityFunction;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -36,13 +39,18 @@ import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
+import org.apache.ignite.internal.processors.cache.distributed.dht.ClientCacheDhtTopologyFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridClientPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAssignmentFetchFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -68,6 +76,10 @@ import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdapter<K, V> {
+ /** */
+ private final long clientCacheMsgTimeout =
+ IgniteSystemProperties.getLong(IgniteSystemProperties.IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT, 10_000);
+
/** Late affinity assignment flag. */
private boolean lateAffAssign;
@@ -81,7 +93,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
private AffinityTopologyVersion lastAffVer;
/** Registered caches (updated from exchange thread). */
- private final Map<Integer, CacheGroupDescriptor> registeredGrps = new HashMap<>();
+ private final CachesInfo caches = new CachesInfo();
/** */
private WaitRebalanceInfo waitInfo;
@@ -93,6 +105,9 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
private final ConcurrentMap<Long, GridDhtAssignmentFetchFuture> pendingAssignmentFetchFuts =
new ConcurrentHashMap8<>();
+ /** */
+ private final ThreadLocal<ClientCacheChangeDiscoveryMessage> clientCacheChanges = new ThreadLocal<>();
+
/** Discovery listener. */
private final GridLocalEventListener discoLsnr = new GridLocalEventListener() {
@Override public void onEvent(Event evt) {
@@ -126,14 +141,13 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
void onDiscoveryEvent(int type, ClusterNode node, AffinityTopologyVersion topVer) {
if (type == EVT_NODE_JOINED && node.isLocal()) {
// Clean-up in case of client reconnect.
- registeredGrps.clear();
+ caches.clear();
affCalcVer = null;
lastAffVer = null;
- for (CacheGroupDescriptor desc : cctx.cache().cacheGroupDescriptors().values())
- registeredGrps.put(desc.groupId(), desc);
+ caches.init(cctx.cache().cacheGroupDescriptors(), cctx.cache().cacheDescriptors());
}
if (!CU.clientNode(node) && (type == EVT_NODE_FAILED || type == EVT_NODE_JOINED || type == EVT_NODE_LEFT)) {
@@ -319,20 +333,312 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
/**
- * @param exchActions Cache change requests to execute on exchange.
+ * @param reqId Request ID.
+ * @param startReqs Client cache start request.
+ * @return Descriptors for caches to start.
*/
- private void updateCachesInfo(ExchangeActions exchActions) {
- for (CacheGroupDescriptor stopDesc : exchActions.cacheGroupsToStop()) {
- CacheGroupDescriptor rmvd = registeredGrps.remove(stopDesc.groupId());
+ @Nullable private List<DynamicCacheDescriptor> clientCachesToStart(UUID reqId,
+ Map<String, DynamicCacheChangeRequest> startReqs) {
+ List<DynamicCacheDescriptor> startDescs = new ArrayList<>(startReqs.size());
+
+ for (DynamicCacheChangeRequest startReq : startReqs.values()) {
+ DynamicCacheDescriptor desc = caches.cache(CU.cacheId(startReq.cacheName()));
+
+ if (desc == null) {
+ CacheException err = new CacheException("Failed to start client cache " +
+ "(a cache with the given name is not started): " + startReq.cacheName());
+
+ cctx.cache().completeClientCacheChangeFuture(reqId, err);
- assert rmvd != null : stopDesc.cacheOrGroupName();
+ return null;
+ }
+
+ if (cctx.cacheContext(desc.cacheId()) != null)
+ continue;
+
+ startDescs.add(desc);
}
- for (CacheGroupDescriptor startDesc : exchActions.cacheGroupsToStart()) {
- CacheGroupDescriptor old = registeredGrps.put(startDesc.groupId(), startDesc);
+ return startDescs;
+ }
- assert old == null : old;
+ /**
+ * @param msg Change request.
+ * @param crd Coordinator flag.
+ * @param topVer Current topology version.
+ * @param discoCache Discovery data cache.
+ * @return Map of started caches (cache ID to near enabled flag).
+ */
+ @Nullable private Map<Integer, Boolean> processClientCacheStartRequests(
+ ClientCacheChangeDummyDiscoveryMessage msg,
+ boolean crd,
+ AffinityTopologyVersion topVer,
+ DiscoCache discoCache) {
+ Map<String, DynamicCacheChangeRequest> startReqs = msg.startRequests();
+
+ if (startReqs == null)
+ return null;
+
+ List<DynamicCacheDescriptor> startDescs = clientCachesToStart(msg.requestId(), msg.startRequests());
+
+ if (startDescs == null || startDescs.isEmpty()) {
+ cctx.cache().completeClientCacheChangeFuture(msg.requestId(), null);
+
+ return null;
+ }
+
+ Map<Integer, GridDhtAssignmentFetchFuture> fetchFuts = U.newHashMap(startDescs.size());
+
+ Set<String> startedCaches = U.newHashSet(startDescs.size());
+
+ Map<Integer, Boolean> startedInfos = U.newHashMap(startDescs.size());
+
+ for (DynamicCacheDescriptor desc : startDescs) {
+ try {
+ startedCaches.add(desc.cacheName());
+
+ DynamicCacheChangeRequest startReq = startReqs.get(desc.cacheName());
+
+ cctx.cache().prepareCacheStart(desc, startReq.nearCacheConfiguration(), topVer);
+
+ startedInfos.put(desc.cacheId(), startReq.nearCacheConfiguration() != null);
+
+ CacheGroupContext grp = cctx.cache().cacheGroup(desc.groupId());
+
+ assert grp != null : desc.groupId();
+ assert !grp.affinityNode() || grp.isLocal() : grp.cacheOrGroupName();
+
+ if (!grp.isLocal() && grp.affinity().lastVersion().equals(AffinityTopologyVersion.NONE)) {
+ assert grp.localStartVersion().equals(topVer) : grp.localStartVersion();
+
+ if (crd) {
+ CacheGroupHolder grpHolder = grpHolders.get(grp.groupId());
+
+ assert grpHolder != null && grpHolder.affinity().idealAssignment() != null;
+
+ if (grpHolder.client()) {
+ ClientCacheDhtTopologyFuture topFut = new ClientCacheDhtTopologyFuture(topVer);
+
+ grp.topology().updateTopologyVersion(topFut, discoCache, -1, false);
+
+ GridClientPartitionTopology clientTop = cctx.exchange().clearClientTopology(grp.groupId());
+
+ if (clientTop != null) {
+ grp.topology().update(topVer,
+ clientTop.partitionMap(true),
+ clientTop.updateCounters(false));
+ }
+
+ grpHolder = new CacheGroupHolder1(grp, grpHolder.affinity());
+
+ grpHolders.put(grp.groupId(), grpHolder);
+
+ assert grpHolder.affinity().lastVersion().equals(grp.affinity().lastVersion());
+ }
+ }
+ else if (!fetchFuts.containsKey(grp.groupId())) {
+ GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
+ grp.groupId(),
+ topVer,
+ discoCache);
+
+ fetchFut.init(true);
+
+ fetchFuts.put(grp.groupId(), fetchFut);
+ }
+ }
+ }
+ catch (IgniteCheckedException e) {
+ cctx.cache().closeCaches(startedCaches, false);
+
+ cctx.cache().completeClientCacheChangeFuture(msg.requestId(), e);
+
+ return null;
+ }
}
+
+ for (GridDhtAssignmentFetchFuture fetchFut : fetchFuts.values()) {
+ try {
+ CacheGroupContext grp = cctx.cache().cacheGroup(fetchFut.groupId());
+
+ assert grp != null;
+
+ GridDhtAffinityAssignmentResponse res = fetchAffinity(topVer,
+ null,
+ discoCache,
+ grp.affinity(),
+ fetchFut);
+
+ GridDhtPartitionFullMap partMap;
+ ClientCacheDhtTopologyFuture topFut;
+
+ if (res != null) {
+ partMap = res.partitionMap();
+
+ assert partMap != null : res;
+
+ topFut = new ClientCacheDhtTopologyFuture(topVer);
+ }
+ else {
+ partMap = new GridDhtPartitionFullMap(cctx.localNodeId(), cctx.localNode().order(), 1);
+
+ topFut = new ClientCacheDhtTopologyFuture(topVer,
+ new ClusterTopologyServerNotFoundException("All server nodes left grid."));
+ }
+
+ grp.topology().updateTopologyVersion(topFut, discoCache, -1, false);
+
+ grp.topology().update(topVer, partMap, null);
+
+ topFut.validate(grp, discoCache.allNodes());
+ }
+ catch (IgniteCheckedException e) {
+ cctx.cache().closeCaches(startedCaches, false);
+
+ cctx.cache().completeClientCacheChangeFuture(msg.requestId(), e);
+
+ return null;
+ }
+ }
+
+ cctx.cache().initCacheProxies(topVer, null);
+
+ cctx.cache().completeClientCacheChangeFuture(msg.requestId(), null);
+
+ return startedInfos;
+ }
+
+ /**
+ * @param msg Change request.
+ * @param topVer Current topology version.
+ * @param crd Coordinator flag.
+ * @return Closed caches IDs.
+ */
+ private Set<Integer> processCacheCloseRequests(
+ ClientCacheChangeDummyDiscoveryMessage msg,
+ boolean crd,
+ AffinityTopologyVersion topVer) {
+ Set<String> cachesToClose = msg.cachesToClose();
+
+ if (cachesToClose == null)
+ return null;
+
+ Set<Integer> closed = cctx.cache().closeCaches(cachesToClose, true);
+
+ if (crd) {
+ for (CacheGroupHolder hld : grpHolders.values()) {
+ if (!hld.client() && cctx.cache().cacheGroup(hld.groupId()) == null) {
+ int grpId = hld.groupId();
+
+ // All client cache groups were stopped, need create 'client' CacheGroupHolder.
+ CacheGroupHolder grpHolder = grpHolders.remove(grpId);
+
+ assert grpHolder != null && !grpHolder.client() : grpHolder;
+
+ try {
+ grpHolder = CacheGroupHolder2.create(cctx,
+ caches.group(grpId),
+ topVer,
+ grpHolder.affinity());
+
+ grpHolders.put(grpId, grpHolder);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to initialize cache: " + e, e);
+ }
+ }
+ }
+ }
+
+ cctx.cache().completeClientCacheChangeFuture(msg.requestId(), null);
+
+ return closed;
+ }
+
+ /**
+ * Process client cache start/close requests, called from exchange thread.
+ *
+ * @param msg Change request.
+ */
+ void processClientCachesChanges(ClientCacheChangeDummyDiscoveryMessage msg) {
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+
+ DiscoCache discoCache = cctx.discovery().discoCache(topVer);
+
+ boolean crd = cctx.localNode().equals(discoCache.oldestAliveServerNode());
+
+ Map<Integer, Boolean> startedCaches = processClientCacheStartRequests(msg, crd, topVer, discoCache);
+
+ Set<Integer> closedCaches = processCacheCloseRequests(msg, crd, topVer);
+
+ if (startedCaches != null || closedCaches != null)
+ scheduleClientChangeMessage(startedCaches, closedCaches);
+ }
+
+ /**
+ * Sends discovery message about started/closed client caches, called from exchange thread.
+ *
+ * @param timeoutObj Timeout object initiated send.
+ */
+ void sendClientCacheChangesMessage(ClientCacheUpdateTimeout timeoutObj) {
+ ClientCacheChangeDiscoveryMessage msg = clientCacheChanges.get();
+
+ // Timeout object was changed if one more client cache changed during timeout,
+ // another timeoutObj was scheduled.
+ if (msg != null && msg.updateTimeoutObject() == timeoutObj) {
+ assert !msg.empty() : msg;
+
+ clientCacheChanges.remove();
+
+ msg.checkCachesExist(caches.registeredCaches.keySet());
+
+ try {
+ if (!msg.empty())
+ cctx.discovery().sendCustomEvent(msg);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to send discovery event: " + e, e);
+ }
+ }
+ }
+
+ /**
+ * @param startedCaches Started caches.
+ * @param closedCaches Closed caches.
+ */
+ private void scheduleClientChangeMessage(Map<Integer, Boolean> startedCaches, Set<Integer> closedCaches) {
+ ClientCacheChangeDiscoveryMessage msg = clientCacheChanges.get();
+
+ if (msg == null) {
+ msg = new ClientCacheChangeDiscoveryMessage(startedCaches, closedCaches);
+
+ clientCacheChanges.set(msg);
+ }
+ else {
+ msg.merge(startedCaches, closedCaches);
+
+ if (msg.empty()) {
+ cctx.time().removeTimeoutObject(msg.updateTimeoutObject());
+
+ clientCacheChanges.remove();
+
+ return;
+ }
+ }
+
+ if (msg.updateTimeoutObject() != null)
+ cctx.time().removeTimeoutObject(msg.updateTimeoutObject());
+
+ long timeout = clientCacheMsgTimeout;
+
+ if (timeout <= 0)
+ timeout = 10_000;
+
+ ClientCacheUpdateTimeout timeoutObj = new ClientCacheUpdateTimeout(cctx, timeout);
+
+ msg.updateTimeoutObject(timeoutObj);
+
+ cctx.time().addTimeoutObject(timeoutObj);
}
/**
@@ -342,16 +648,15 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
* @param crd Coordinator flag.
* @param exchActions Cache change requests.
* @throws IgniteCheckedException If failed.
- * @return {@code True} if client-only exchange is needed.
*/
- public boolean onCacheChangeRequest(final GridDhtPartitionsExchangeFuture fut,
+ public void onCacheChangeRequest(final GridDhtPartitionsExchangeFuture fut,
boolean crd,
final ExchangeActions exchActions)
throws IgniteCheckedException
{
assert exchActions != null && !exchActions.empty() : exchActions;
- updateCachesInfo(exchActions);
+ caches.updateCachesInfo(exchActions);
// Affinity did not change for existing caches.
forAllCacheGroups(crd && lateAffAssign, new IgniteInClosureX<GridAffinityAssignmentCache>() {
@@ -363,7 +668,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
});
- for (ExchangeActions.ActionData action : exchActions.newAndClientCachesStartRequests()) {
+ for (ExchangeActions.ActionData action : exchActions.cacheStartRequests()) {
DynamicCacheDescriptor cacheDesc = action.descriptor();
DynamicCacheChangeRequest req = action.request();
@@ -396,81 +701,34 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
U.error(log, "Failed to initialize cache. Will try to rollback cache start routine. " +
"[cacheName=" + req.cacheName() + ']', e);
- cctx.cache().forceCloseCache(fut.topologyVersion(), action, e);
- }
- }
-
- Set<Integer> gprs = new HashSet<>();
-
- for (ExchangeActions.ActionData action : exchActions.newAndClientCachesStartRequests()) {
- Integer grpId = action.descriptor().groupId();
+ cctx.cache().closeCaches(Collections.singleton(req.cacheName()), false);
- if (gprs.add(grpId)) {
- if (crd && lateAffAssign)
- initStartedGroupOnCoordinator(fut, action.descriptor().groupDescriptor());else {
- CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
-
- if (grp != null && !grp.isLocal() && grp.localStartVersion().equals(fut.topologyVersion())) {
- assert grp.affinity().lastVersion().equals(AffinityTopologyVersion.NONE) : grp.affinity().lastVersion();
-
- initAffinity(registeredGrps.get(grp.groupId()), grp.affinity(), fut);
- }
- }
+ cctx.cache().completeCacheStartFuture(req, false, e);
}
}
- List<ExchangeActions.ActionData> closeReqs = exchActions.closeRequests(cctx.localNodeId());
-
- for (ExchangeActions.ActionData req : closeReqs) {
- cctx.cache().blockGateway(req.request());
-
- if (crd) {
- CacheGroupContext grp = cctx.cache().cacheGroup(req.descriptor().groupId());
-
- assert grp != null;
-
- if (grp.affinityNode())
- continue;
-
- boolean grpClosed = false;
-
- if (grp.sharedGroup()) {
- boolean cacheRemaining = false;
-
- for (GridCacheContext ctx : cctx.cacheContexts()) {
- if (ctx.group() == grp && !cacheClosed(ctx.cacheId(), closeReqs)) {
- cacheRemaining = true;
-
- break;
- }
- }
-
- if (!cacheRemaining)
- grpClosed = true;
- }
- else
- grpClosed = true;
+ Set<Integer> gprs = new HashSet<>();
- // All client cache groups were stopped, need create 'client' CacheGroupHolder.
- if (grpClosed) {
- CacheGroupHolder grpHolder = grpHolders.remove(grp.groupId());
+ for (ExchangeActions.ActionData action : exchActions.cacheStartRequests()) {
+ Integer grpId = action.descriptor().groupId();
- if (grpHolder != null) {
- assert !grpHolder.client() : grpHolder;
+ if (gprs.add(grpId)) {
+ if (crd && lateAffAssign)
+ initStartedGroupOnCoordinator(fut, action.descriptor().groupDescriptor());
+ else {
+ CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
- grpHolder = CacheGroupHolder2.create(cctx,
- registeredGrps.get(grp.groupId()),
- fut,
- grp.affinity());
+ if (grp != null && !grp.isLocal() && grp.localStartVersion().equals(fut.topologyVersion())) {
+ assert grp.affinity().lastVersion().equals(AffinityTopologyVersion.NONE) : grp.affinity().lastVersion();
- grpHolders.put(grp.groupId(), grpHolder);
+ initAffinity(caches.group(grp.groupId()), grp.affinity(), fut);
}
}
}
}
for (ExchangeActions.ActionData action : exchActions.cacheStopRequests())
- cctx.cache().blockGateway(action.request());
+ cctx.cache().blockGateway(action.request().cacheName(), true);
Set<Integer> stoppedGrps = null;
@@ -519,21 +777,14 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
}
- return exchActions.clientOnlyExchange();
- }
+ ClientCacheChangeDiscoveryMessage msg = clientCacheChanges.get();
- /**
- * @param cacheId Cache ID.
- * @param closeReqs Close requests.
- * @return {@code True} if requests contain request for given cache ID.
- */
- private boolean cacheClosed(int cacheId, List<ExchangeActions.ActionData> closeReqs) {
- for (ExchangeActions.ActionData req : closeReqs) {
- if (req.descriptor().cacheId() == cacheId)
- return true;
- }
+ if (msg != null) {
+ msg.checkCachesExist(caches.registeredCaches.keySet());
- return false;
+ if (msg.empty())
+ clientCacheChanges.remove();
+ }
}
/**
@@ -542,7 +793,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
public void removeAllCacheInfo(){
grpHolders.clear();
- registeredGrps.clear();
+ caches.clear();
}
/**
@@ -633,7 +884,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
assert affTopVer.topologyVersion() > 0 : affTopVer;
- CacheGroupDescriptor desc = registeredGrps.get(aff.groupId());
+ CacheGroupDescriptor desc = caches.group(aff.groupId());
assert desc != null : aff.cacheOrGroupName();
@@ -763,7 +1014,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
private void forAllRegisteredCacheGroups(IgniteInClosureX<CacheGroupDescriptor> c) throws IgniteCheckedException {
assert lateAffAssign;
- for (CacheGroupDescriptor cacheDesc : registeredGrps.values()) {
+ for (CacheGroupDescriptor cacheDesc : caches.allGroups()) {
if (cacheDesc.config().getCacheMode() == LOCAL)
continue;
@@ -811,7 +1062,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
if (grpHolder == null) {
grpHolder = grp != null ?
new CacheGroupHolder1(grp, null) :
- CacheGroupHolder2.create(cctx, grpDesc, fut, null);
+ CacheGroupHolder2.create(cctx, grpDesc, fut.topologyVersion(), null);
CacheGroupHolder old = grpHolders.put(grpId, grpHolder);
@@ -843,12 +1094,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
public void initStartedCaches(boolean crd,
final GridDhtPartitionsExchangeFuture fut,
Collection<DynamicCacheDescriptor> descs) throws IgniteCheckedException {
- for (DynamicCacheDescriptor desc : descs) {
- CacheGroupDescriptor grpDesc = desc.groupDescriptor();
-
- if (!registeredGrps.containsKey(grpDesc.groupId()))
- registeredGrps.put(grpDesc.groupId(), grpDesc);
- }
+ caches.initStartedCaches(descs);
if (crd && lateAffAssign) {
forAllRegisteredCacheGroups(new IgniteInClosureX<CacheGroupDescriptor>() {
@@ -868,7 +1114,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
forAllCacheGroups(false, new IgniteInClosureX<GridAffinityAssignmentCache>() {
@Override public void applyx(GridAffinityAssignmentCache aff) throws IgniteCheckedException {
if (aff.lastVersion().equals(AffinityTopologyVersion.NONE))
- initAffinity(registeredGrps.get(aff.groupId()), aff, fut);
+ initAffinity(caches.group(aff.groupId()), aff, fut);
}
});
}
@@ -893,13 +1139,16 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
else {
GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
- desc,
+ desc.groupId(),
fut.topologyVersion(),
fut.discoCache());
- fetchFut.init();
+ fetchFut.init(false);
- fetchAffinity(fut, aff, fetchFut);
+ fetchAffinity(fut.topologyVersion(),
+ fut.discoveryEvent(),
+ fut.discoCache(),
+ aff, fetchFut);
}
}
@@ -990,7 +1239,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
StringBuilder names = new StringBuilder();
for (Integer grpId : grpIds) {
- String name = registeredGrps.get(grpId).cacheOrGroupName();
+ String name = caches.group(grpId).cacheOrGroupName();
if (names.length() != 0)
names.append(", ");
@@ -1022,16 +1271,16 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
grp.affinity().initialize(fut.topologyVersion(), assignment);
}
else {
- CacheGroupDescriptor grpDesc = registeredGrps.get(grp.groupId());
+ CacheGroupDescriptor grpDesc = caches.group(grp.groupId());
assert grpDesc != null : grp.cacheOrGroupName();
GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
- grpDesc,
+ grpDesc.groupId(),
topVer,
fut.discoCache());
- fetchFut.init();
+ fetchFut.init(false);
fetchFuts.add(fetchFut);
}
@@ -1042,48 +1291,57 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
Integer grpId = fetchFut.groupId();
- fetchAffinity(fut, cctx.cache().cacheGroup(grpId).affinity(), fetchFut);
+ fetchAffinity(fut.topologyVersion(),
+ fut.discoveryEvent(),
+ fut.discoCache(),
+ cctx.cache().cacheGroup(grpId).affinity(),
+ fetchFut);
}
}
/**
- * @param fut Exchange future.
+ * @param topVer Topology version.
+ * @param discoveryEvt Discovery event.
+ * @param discoCache Discovery data cache.
* @param affCache Affinity.
* @param fetchFut Affinity fetch future.
* @throws IgniteCheckedException If failed.
+ * @return Affinity assignment response.
*/
- private void fetchAffinity(GridDhtPartitionsExchangeFuture fut,
+ private GridDhtAffinityAssignmentResponse fetchAffinity(AffinityTopologyVersion topVer,
+ @Nullable DiscoveryEvent discoveryEvt,
+ DiscoCache discoCache,
GridAffinityAssignmentCache affCache,
GridDhtAssignmentFetchFuture fetchFut)
throws IgniteCheckedException {
assert affCache != null;
- AffinityTopologyVersion topVer = fut.topologyVersion();
-
GridDhtAffinityAssignmentResponse res = fetchFut.get();
if (res == null) {
- List<List<ClusterNode>> aff = affCache.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
+ List<List<ClusterNode>> aff = affCache.calculate(topVer, discoveryEvt, discoCache);
affCache.initialize(topVer, aff);
}
else {
- List<List<ClusterNode>> idealAff = res.idealAffinityAssignment(cctx.discovery());
+ List<List<ClusterNode>> idealAff = res.idealAffinityAssignment(discoCache);
if (idealAff != null)
affCache.idealAssignment(idealAff);
else {
assert !affCache.centralizedAffinityFunction() || !lateAffAssign;
- affCache.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
+ affCache.calculate(topVer, discoveryEvt, discoCache);
}
- List<List<ClusterNode>> aff = res.affinityAssignment(cctx.discovery());
+ List<List<ClusterNode>> aff = res.affinityAssignment(discoCache);
assert aff != null : res;
affCache.initialize(topVer, aff);
}
+
+ return res;
}
/**
@@ -1136,7 +1394,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
if (grp.isLocal())
continue;
- initAffinity(registeredGrps.get(grp.groupId()), grp.affinity(), fut);
+ initAffinity(caches.group(grp.groupId()), grp.affinity(), fut);
}
}
@@ -1176,7 +1434,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
);
- grpHolder = CacheGroupHolder2.create(cctx, desc, fut, null);
+ grpHolder = CacheGroupHolder2.create(cctx, desc, fut.topologyVersion(), null);
final GridAffinityAssignmentCache aff = grpHolder.affinity();
@@ -1198,18 +1456,21 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
assert prev.topologyVersion().compareTo(fut.topologyVersion()) < 0 : prev;
GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
- desc,
+ desc.groupId(),
prev.topologyVersion(),
prev.discoCache());
- fetchFut.init();
+ fetchFut.init(false);
final GridFutureAdapter<AffinityTopologyVersion> affFut = new GridFutureAdapter<>();
fetchFut.listen(new IgniteInClosureX<IgniteInternalFuture<GridDhtAffinityAssignmentResponse>>() {
@Override public void applyx(IgniteInternalFuture<GridDhtAffinityAssignmentResponse> fetchFut)
throws IgniteCheckedException {
- fetchAffinity(prev, aff, (GridDhtAssignmentFetchFuture)fetchFut);
+ fetchAffinity(prev.topologyVersion(),
+ prev.discoveryEvent(),
+ prev.discoCache(),
+ aff, (GridDhtAssignmentFetchFuture)fetchFut);
aff.calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
@@ -1268,7 +1529,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
);
- cacheGrp = CacheGroupHolder2.create(cctx, desc, fut, null);
+ cacheGrp = CacheGroupHolder2.create(cctx, desc, fut.topologyVersion(), null);
}
else
cacheGrp = new CacheGroupHolder1(grp, null);
@@ -1749,7 +2010,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
/**
* @param cctx Context.
* @param grpDesc Cache group descriptor.
- * @param fut Exchange future.
+ * @param topVer Current exchange version.
* @param initAff Current affinity.
* @return Cache holder.
* @throws IgniteCheckedException If failed.
@@ -1757,7 +2018,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
static CacheGroupHolder2 create(
GridCacheSharedContext cctx,
CacheGroupDescriptor grpDesc,
- GridDhtPartitionsExchangeFuture fut,
+ AffinityTopologyVersion topVer,
@Nullable GridAffinityAssignmentCache initAff) throws IgniteCheckedException {
assert grpDesc != null;
assert !cctx.kernalContext().clientNode();
@@ -1768,7 +2029,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
assert ccfg.getCacheMode() != LOCAL : ccfg.getName();
assert !cctx.discovery().cacheGroupAffinityNodes(grpDesc.groupId(),
- fut.topologyVersion()).contains(cctx.localNode()) : grpDesc.cacheOrGroupName();
+ topVer).contains(cctx.localNode()) : grpDesc.cacheOrGroupName();
AffinityFunction affFunc = cctx.cache().clone(ccfg.getAffinity());
@@ -1872,7 +2133,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
if (cacheWaitParts == null) {
waitGrps.put(grpId, cacheWaitParts = new HashMap<>());
- deploymentIds.put(grpId, registeredGrps.get(grpId).deploymentId());
+ deploymentIds.put(grpId, caches.group(grpId).deploymentId());
}
cacheWaitParts.put(part, waitNode);
@@ -1891,4 +2152,101 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
", grps=" + (waitGrps != null ? waitGrps.keySet() : null) + ']';
}
}
+
+ /**
+ *
+ */
+ static class CachesInfo {
+ /** Registered cache groups (updated from exchange thread). */
+ private final ConcurrentHashMap<Integer, CacheGroupDescriptor> registeredGrps = new ConcurrentHashMap<>();
+
+ /** Registered caches (updated from exchange thread). */
+ private final ConcurrentHashMap<Integer, DynamicCacheDescriptor> registeredCaches = new ConcurrentHashMap<>();
+
+ /**
+ * @param grps Registered groups.
+ * @param caches Registered caches.
+ */
+ void init(Map<Integer, CacheGroupDescriptor> grps, Map<String, DynamicCacheDescriptor> caches) {
+ for (CacheGroupDescriptor grpDesc : grps.values())
+ registeredGrps.put(grpDesc.groupId(), grpDesc);
+
+ for (DynamicCacheDescriptor cacheDesc : caches.values())
+ registeredCaches.put(cacheDesc.cacheId(), cacheDesc);
+ }
+
+ /**
+ * @return All registered groups.
+ */
+ Collection<CacheGroupDescriptor> allGroups() {
+ return registeredGrps.values();
+ }
+
+ /**
+ * @param grpId Group ID.
+ * @return Group descriptor.
+ */
+ CacheGroupDescriptor group(int grpId) {
+ CacheGroupDescriptor desc = registeredGrps.get(grpId);
+
+ assert desc != null : grpId;
+
+ return desc;
+ }
+
+ /**
+ * @param descs Cache descriptor.
+ */
+ void initStartedCaches(Collection<DynamicCacheDescriptor> descs) {
+ for (DynamicCacheDescriptor desc : descs) {
+ CacheGroupDescriptor grpDesc = desc.groupDescriptor();
+
+ if (!registeredGrps.containsKey(grpDesc.groupId()))
+ registeredGrps.put(grpDesc.groupId(), grpDesc);
+
+ if (!registeredCaches.containsKey(desc.cacheName()))
+ registeredCaches.put(desc.cacheId(), desc);
+ }
+ }
+
+ /**
+ * @param exchActions Exchange actions.
+ */
+ void updateCachesInfo(ExchangeActions exchActions) {
+ for (CacheGroupDescriptor stopDesc : exchActions.cacheGroupsToStop()) {
+ CacheGroupDescriptor rmvd = registeredGrps.remove(stopDesc.groupId());
+
+ assert rmvd != null : stopDesc.cacheOrGroupName();
+ }
+
+ for (CacheGroupDescriptor startDesc : exchActions.cacheGroupsToStart()) {
+ CacheGroupDescriptor old = registeredGrps.put(startDesc.groupId(), startDesc);
+
+ assert old == null : old;
+ }
+
+ for (ExchangeActions.ActionData req : exchActions.cacheStopRequests())
+ registeredCaches.remove(req.descriptor().cacheId());
+
+ for (ExchangeActions.ActionData req : exchActions.cacheStartRequests())
+ registeredCaches.put(req.descriptor().cacheId(), req.descriptor());
+ }
+
+ /**
+ * @param cacheId Cache ID.
+ * @return Cache descriptor if cache found.
+ */
+ @Nullable DynamicCacheDescriptor cache(Integer cacheId) {
+ return registeredCaches.get(cacheId);
+ }
+
+ /**
+ *
+ */
+ void clear() {
+ registeredGrps.clear();
+
+ registeredCaches.clear();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
index 4844a55..196df57 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
@@ -898,6 +898,9 @@ public class CacheGroupContext {
res.idealAffinityAssignment(assignment.idealAssignment());
}
+ if (req.sendPartitionsState())
+ res.partitionMap(top.partitionMap(true));
+
try {
ctx.io().send(nodeId, res, AFFINITY_POOL);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDiscoveryMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDiscoveryMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDiscoveryMessage.java
new file mode 100644
index 0000000..3d120f7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDiscoveryMessage.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Sent from cache client node to asynchronously notify about started.closed client caches.
+ */
+public class ClientCacheChangeDiscoveryMessage implements DiscoveryCustomMessage {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final IgniteUuid id = IgniteUuid.randomUuid();;
+
+ /** */
+ @GridToStringInclude
+ private Map<Integer, Boolean> startedCaches;
+
+ /** */
+ @GridToStringInclude
+ private Set<Integer> closedCaches;
+
+ /** Update timeout object, used to batch multiple starts/close into single discovery message. */
+ private transient ClientCacheUpdateTimeout updateTimeoutObj;
+
+ /**
+ * @param startedCaches Started caches.
+ * @param closedCaches Closed caches.
+ */
+ public ClientCacheChangeDiscoveryMessage(Map<Integer, Boolean> startedCaches, Set<Integer> closedCaches) {
+ this.startedCaches = startedCaches;
+ this.closedCaches = closedCaches;
+ }
+
+ /**
+ * @param startedCaches Started caches.
+ * @param closedCaches Closed caches.
+ */
+ public void merge(@Nullable Map<Integer, Boolean> startedCaches, @Nullable Set<Integer> closedCaches) {
+ Map<Integer, Boolean> startedCaches0 = this.startedCaches;
+ Set<Integer> closedCaches0 = this.closedCaches;
+
+ if (startedCaches != null) {
+ if (startedCaches0 == null)
+ this.startedCaches = startedCaches0 = new HashMap<>();
+
+ for (Map.Entry<Integer, Boolean> e : startedCaches.entrySet()) {
+ if (closedCaches0 != null && closedCaches0.remove(e.getKey()))
+ continue;
+
+ Boolean old = startedCaches0.put(e.getKey(), e.getValue());
+
+ assert old == null : e.getKey();
+ }
+ }
+
+ if (closedCaches != null) {
+ if (closedCaches0 == null)
+ this.closedCaches = closedCaches0 = new HashSet<>();
+
+ for (Integer cacheId : closedCaches) {
+ if (startedCaches0 != null && startedCaches0.remove(cacheId) != null)
+ continue;
+
+ boolean add = closedCaches0.add(cacheId);
+
+ assert add : cacheId;
+ }
+ }
+ }
+
+ /**
+ * @return {@code True} if there are no info about started/closed caches.
+ */
+ public boolean empty() {
+ return F.isEmpty(startedCaches) && F.isEmpty(closedCaches);
+ }
+
+ /**
+ * @param caches Started caches' IDs.
+ */
+ void checkCachesExist(Set<Integer> caches) {
+ if (closedCaches != null) {
+ for (Iterator<Integer> it = closedCaches.iterator(); it.hasNext();) {
+ Integer cacheId = it.next();
+
+ if (!caches.contains(cacheId))
+ it.remove();
+ }
+ }
+
+ if (startedCaches != null) {
+ for (Iterator<Integer> it = startedCaches.keySet().iterator(); it.hasNext();) {
+ Integer cacheId = it.next();
+
+ if (!caches.contains(cacheId))
+ it.remove();
+ }
+ }
+ }
+
+ /**
+ * @return Update timeout object.
+ */
+ public ClientCacheUpdateTimeout updateTimeoutObject() {
+ return updateTimeoutObj;
+ }
+
+ /**
+ * @param updateTimeoutObj Update timeout object.
+ */
+ public void updateTimeoutObject(ClientCacheUpdateTimeout updateTimeoutObj) {
+ this.updateTimeoutObj = updateTimeoutObj;
+ }
+
+ /**
+ * @return Started caches map (cache ID to near enabled flag).
+ */
+ @Nullable public Map<Integer, Boolean> startedCaches() {
+ return startedCaches;
+ }
+
+ /**
+ * @return Closed caches.
+ */
+ @Nullable public Set<Integer> closedCaches() {
+ return closedCaches;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid id() {
+ return id;
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isMutable() {
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(ClientCacheChangeDiscoveryMessage.class, this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDummyDiscoveryMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDummyDiscoveryMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDummyDiscoveryMessage.java
new file mode 100644
index 0000000..68bca27
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheChangeDummyDiscoveryMessage.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteUuid;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Dummy discovery message which is not really sent via ring, it is just added in local discovery worker queue.
+ */
+public class ClientCacheChangeDummyDiscoveryMessage implements DiscoveryCustomMessage,
+ CachePartitionExchangeWorkerTask {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final UUID reqId;
+
+ /** */
+ private final Map<String, DynamicCacheChangeRequest> startReqs;
+
+ /** */
+ @GridToStringInclude
+ private final Set<String> cachesToClose;
+
+ /**
+ * @param reqId Start request ID.
+ * @param startReqs Caches start requests.
+ * @param cachesToClose Cache to close.
+ */
+ public ClientCacheChangeDummyDiscoveryMessage(UUID reqId,
+ @Nullable Map<String, DynamicCacheChangeRequest> startReqs,
+ @Nullable Set<String> cachesToClose) {
+ assert reqId != null;
+ assert startReqs != null ^ cachesToClose != null;
+
+ this.reqId = reqId;
+ this.startReqs = startReqs;
+ this.cachesToClose = cachesToClose;
+ }
+
+ /**
+ * @return Start request ID.
+ */
+ UUID requestId() {
+ return reqId;
+ }
+
+ /**
+ * @return Cache start requests.
+ */
+ @Nullable Map<String, DynamicCacheChangeRequest> startRequests() {
+ return startReqs;
+ }
+
+ /**
+ * @return Client caches to close.
+ */
+ Set<String> cachesToClose() {
+ return cachesToClose;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid id() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public DiscoveryCustomMessage ackMessage() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean isMutable() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(ClientCacheChangeDummyDiscoveryMessage.class, this,
+ "startCaches", (startReqs != null ? startReqs.keySet() : ""));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheUpdateTimeout.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheUpdateTimeout.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheUpdateTimeout.java
new file mode 100644
index 0000000..aab3a3e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClientCacheUpdateTimeout.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
+
+/**
+ *
+ */
+class ClientCacheUpdateTimeout extends GridTimeoutObjectAdapter implements CachePartitionExchangeWorkerTask {
+ /** */
+ private final GridCacheSharedContext cctx;
+
+ /**
+ * @param cctx Context.
+ * @param timeout Timeout.
+ */
+ ClientCacheUpdateTimeout(GridCacheSharedContext cctx, long timeout) {
+ super(timeout);
+
+ this.cctx = cctx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onTimeout() {
+ if (!cctx.kernalContext().isStopping())
+ cctx.exchange().addCustomTask(this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
index d5718f8..7398074 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
@@ -280,6 +280,40 @@ class ClusterCachesInfo {
}
/**
+ * @param msg Message.
+ * @param node Node sent message.
+ */
+ void onClientCacheChange(ClientCacheChangeDiscoveryMessage msg, ClusterNode node) {
+ Map<Integer, Boolean> startedCaches = msg.startedCaches();
+
+ if (startedCaches != null) {
+ for (Map.Entry<Integer, Boolean> e : startedCaches.entrySet()) {
+ for (DynamicCacheDescriptor desc : registeredCaches.values()) {
+ if (e.getKey().equals(desc.cacheId())) {
+ ctx.discovery().addClientNode(desc.cacheName(), node.id(), e.getValue());
+
+ break;
+ }
+ }
+ }
+ }
+
+ Set<Integer> closedCaches = msg.closedCaches();
+
+ if (closedCaches != null) {
+ for (Integer cacheId : closedCaches) {
+ for (DynamicCacheDescriptor desc : registeredCaches.values()) {
+ if (cacheId.equals(desc.cacheId())) {
+ ctx.discovery().onClientCacheClose(desc.cacheName(), node.id());
+
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ /**
* @param batch Cache change request.
* @param topVer Topology version.
* @return {@code True} if minor topology version should be increased.
@@ -325,10 +359,14 @@ class ClusterCachesInfo {
continue;
}
+ assert !req.clientStartOnly() : req;
+
DynamicCacheDescriptor desc = req.globalStateChange() ? null : registeredCaches.get(req.cacheName());
boolean needExchange = false;
+ boolean clientCacheStart = false;
+
AffinityTopologyVersion waitTopVer = null;
if (req.start()) {
@@ -388,6 +426,7 @@ class ClusterCachesInfo {
assert old == null;
ctx.discovery().setCacheFilter(
+ startDesc.cacheId(),
grpDesc.groupId(),
ccfg.getName(),
ccfg.getNearConfiguration() != null);
@@ -406,40 +445,35 @@ class ClusterCachesInfo {
else {
assert req.initiatingNodeId() != null : req;
- // Cache already exists, exchange is needed only if client cache should be created.
- ClusterNode node = ctx.discovery().node(req.initiatingNodeId());
-
- boolean clientReq = node != null &&
- !ctx.discovery().cacheAffinityNode(node, req.cacheName());
-
- if (req.clientStartOnly()) {
- needExchange = clientReq && ctx.discovery().addClientNode(req.cacheName(),
- req.initiatingNodeId(),
- req.nearCacheConfiguration() != null);
+ if (req.failIfExists()) {
+ ctx.cache().completeCacheStartFuture(req, false,
+ new CacheExistsException("Failed to start cache " +
+ "(a cache with the same name is already started): " + req.cacheName()));
}
else {
- if (req.failIfExists()) {
- ctx.cache().completeCacheStartFuture(req, false,
- new CacheExistsException("Failed to start cache " +
- "(a cache with the same name is already started): " + req.cacheName()));
- }
- else {
- needExchange = clientReq && ctx.discovery().addClientNode(req.cacheName(),
+ // Cache already exists, it is possible client cache is needed.
+ ClusterNode node = ctx.discovery().node(req.initiatingNodeId());
+
+ boolean clientReq = node != null &&
+ !ctx.discovery().cacheAffinityNode(node, req.cacheName());
+
+ if (clientReq) {
+ ctx.discovery().addClientNode(req.cacheName(),
req.initiatingNodeId(),
req.nearCacheConfiguration() != null);
- }
- }
- if (needExchange) {
- req.clientStartOnly(true);
+ if (node.id().equals(req.initiatingNodeId())) {
+ desc.clientCacheStartVersion(topVer);
- desc.clientCacheStartVersion(topVer.nextMinorVersion());
+ clientCacheStart = true;
- exchangeActions.addClientCacheToStart(req, desc);
+ ctx.discovery().clientCacheStartEvent(req.requestId(), F.asMap(req.cacheName(), req), null);
+ }
+ }
}
}
- if (!needExchange && desc != null) {
+ if (!needExchange && !clientCacheStart && desc != null) {
if (desc.clientCacheStartVersion() != null)
waitTopVer = desc.clientCacheStartVersion();
else {
@@ -508,19 +542,11 @@ class ClusterCachesInfo {
}
}
}
- else if (req.close()) {
- if (desc != null) {
- needExchange = ctx.discovery().onClientCacheClose(req.cacheName(), req.initiatingNodeId());
-
- if (needExchange)
- exchangeActions.addCacheToClose(req, desc);
- }
- }
else
assert false : req;
if (!needExchange) {
- if (req.initiatingNodeId().equals(ctx.localNodeId()))
+ if (!clientCacheStart && req.initiatingNodeId().equals(ctx.localNodeId()))
reqsToComplete.add(new T2<>(req, waitTopVer));
}
else
@@ -849,6 +875,7 @@ class ClusterCachesInfo {
registeredCaches.put(cacheData.cacheConfiguration().getName(), desc);
ctx.discovery().setCacheFilter(
+ desc.cacheId(),
grpDesc.groupId(),
cfg.getName(),
cfg.getNearConfiguration() != null);
@@ -1082,6 +1109,7 @@ class ClusterCachesInfo {
joinData.cacheDeploymentId());
ctx.discovery().setCacheFilter(
+ cacheId,
grpDesc.groupId(),
cfg.getName(),
cfg.getNearConfiguration() != null);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
index 5434061..c6da43f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/DynamicCacheChangeRequest.java
@@ -67,9 +67,6 @@ public class DynamicCacheChangeRequest implements Serializable {
/** Destroy. */
private boolean destroy;
- /** Close flag. */
- private boolean close;
-
/** Whether cache was created through SQL. */
private boolean sql;
@@ -155,19 +152,6 @@ public class DynamicCacheChangeRequest implements Serializable {
/**
* @param ctx Context.
* @param cacheName Cache name.
- * @return Request to close client cache.
- */
- static DynamicCacheChangeRequest closeRequest(GridKernalContext ctx, String cacheName) {
- DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(UUID.randomUUID(), cacheName, ctx.localNodeId());
-
- req.close(true);
-
- return req;
- }
-
- /**
- * @param ctx Context.
- * @param cacheName Cache name.
* @param sql {@code true} if the cache must be stopped only if it was created by SQL command {@code CREATE TABLE}.
* @param destroy Destroy flag.
* @return Cache stop request.
@@ -372,20 +356,6 @@ public class DynamicCacheChangeRequest implements Serializable {
}
/**
- * @return Close flag.
- */
- public boolean close() {
- return close;
- }
-
- /**
- * @param close New close flag.
- */
- public void close(boolean close) {
- this.close = close;
- }
-
- /**
* @return SQL flag.
*/
public boolean sql() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeActions.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeActions.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeActions.java
index d577b30..31546be 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeActions.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeActions.java
@@ -43,15 +43,9 @@ public class ExchangeActions {
private Map<String, ActionData> cachesToStart;
/** */
- private Map<String, ActionData> clientCachesToStart;
-
- /** */
private Map<String, ActionData> cachesToStop;
/** */
- private Map<String, ActionData> cachesToClose;
-
- /** */
private Map<String, ActionData> cachesToResetLostParts;
/** */
@@ -60,7 +54,7 @@ public class ExchangeActions {
/**
* @return {@code True} if server nodes should not participate in exchange.
*/
- boolean clientOnlyExchange() {
+ public boolean clientOnlyExchange() {
return F.isEmpty(cachesToStart) &&
F.isEmpty(cachesToStop) &&
F.isEmpty(cacheGrpsToStart) &&
@@ -69,27 +63,6 @@ public class ExchangeActions {
}
/**
- * @param nodeId Local node ID.
- * @return Close cache requests.
- */
- List<ActionData> closeRequests(UUID nodeId) {
- List<ActionData> res = null;
-
- if (cachesToClose != null) {
- for (ActionData req : cachesToClose.values()) {
- if (nodeId.equals(req.req.initiatingNodeId())) {
- if (res == null)
- res = new ArrayList<>(cachesToClose.size());
-
- res.add(req);
- }
- }
- }
-
- return res != null ? res : Collections.<ActionData>emptyList();
- }
-
- /**
* @return New caches start requests.
*/
Collection<ActionData> cacheStartRequests() {
@@ -97,25 +70,6 @@ public class ExchangeActions {
}
/**
- * @return Start cache requests.
- */
- Collection<ActionData> newAndClientCachesStartRequests() {
- if (cachesToStart != null || clientCachesToStart != null) {
- List<ActionData> res = new ArrayList<>();
-
- if (cachesToStart != null)
- res.addAll(cachesToStart.values());
-
- if (clientCachesToStart != null)
- res.addAll(clientCachesToStart.values());
-
- return res;
- }
-
- return Collections.emptyList();
- }
-
- /**
* @return Stop cache requests.
*/
Collection<ActionData> cacheStopRequests() {
@@ -128,8 +82,6 @@ public class ExchangeActions {
public void completeRequestFutures(GridCacheSharedContext ctx) {
completeRequestFutures(cachesToStart, ctx);
completeRequestFutures(cachesToStop, ctx);
- completeRequestFutures(cachesToClose, ctx);
- completeRequestFutures(clientCachesToStart, ctx);
completeRequestFutures(cachesToResetLostParts, ctx);
}
@@ -194,21 +146,6 @@ public class ExchangeActions {
}
/**
- * @param nodeId Local node ID.
- * @return {@code True} if client cache was started.
- */
- public boolean clientCacheStarted(UUID nodeId) {
- if (clientCachesToStart != null) {
- for (ActionData cache : clientCachesToStart.values()) {
- if (nodeId.equals(cache.req.initiatingNodeId()))
- return true;
- }
- }
-
- return false;
- }
-
- /**
* @param state New cluster state.
*/
void newClusterState(ClusterState state) {
@@ -260,16 +197,6 @@ public class ExchangeActions {
* @param req Request.
* @param desc Cache descriptor.
*/
- void addClientCacheToStart(DynamicCacheChangeRequest req, DynamicCacheDescriptor desc) {
- assert req.start() : req;
-
- clientCachesToStart = add(clientCachesToStart, req, desc);
- }
-
- /**
- * @param req Request.
- * @param desc Cache descriptor.
- */
void addCacheToStop(DynamicCacheChangeRequest req, DynamicCacheDescriptor desc) {
assert req.stop() : req;
@@ -280,16 +207,6 @@ public class ExchangeActions {
* @param req Request.
* @param desc Cache descriptor.
*/
- void addCacheToClose(DynamicCacheChangeRequest req, DynamicCacheDescriptor desc) {
- assert req.close() : req;
-
- cachesToClose = add(cachesToClose, req, desc);
- }
-
- /**
- * @param req Request.
- * @param desc Cache descriptor.
- */
void addCacheToResetLostPartitions(DynamicCacheChangeRequest req, DynamicCacheDescriptor desc) {
assert req.resetLostPartitions() : req;
@@ -369,9 +286,7 @@ public class ExchangeActions {
*/
public boolean empty() {
return F.isEmpty(cachesToStart) &&
- F.isEmpty(clientCachesToStart) &&
F.isEmpty(cachesToStop) &&
- F.isEmpty(cachesToClose) &&
F.isEmpty(cacheGrpsToStart) &&
F.isEmpty(cacheGrpsToStop) &&
F.isEmpty(cachesToResetLostParts);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index 839ddbd..d753f99 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -829,14 +829,7 @@ public class GridCacheContext<K, V> implements Externalizable {
* @return Partition topology.
*/
public GridDhtPartitionTopology topology() {
- GridCacheAdapter<K, V> cache = this.cache;
-
- if (cache == null)
- throw new IllegalStateException("Cache stopped: " + cacheName);
-
- assert cache.isNear() || cache.isDht() || cache.isColocated() || cache.isDhtAtomic() : cache;
-
- return topology(cache);
+ return grp.topology();
}
/**
[2/4] ignite git commit: ignite-5272 Do not use synchronous custom
discovery event for client cache start/close
Posted by sb...@apache.org.
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 2271a85..571cc3d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -35,7 +35,6 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.cache.PartitionLossPolicy;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.events.CacheEvent;
import org.apache.ignite.events.DiscoveryEvent;
@@ -57,7 +56,6 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
import org.apache.ignite.internal.processors.cache.CacheAffinityChangeMessage;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
-import org.apache.ignite.internal.processors.cache.CacheInvalidStateException;
import org.apache.ignite.internal.processors.cache.CachePartitionExchangeWorkerTask;
import org.apache.ignite.internal.processors.cache.ClusterState;
import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
@@ -66,11 +64,11 @@ import org.apache.ignite.internal.processors.cache.ExchangeActions;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFutureAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridClientPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cluster.GridClusterStateProcessor;
@@ -95,10 +93,6 @@ import static org.apache.ignite.IgniteSystemProperties.IGNITE_LONG_OPERATIONS_DU
import static org.apache.ignite.IgniteSystemProperties.IGNITE_THREAD_DUMP_ON_EXCHANGE_TIMEOUT;
import static org.apache.ignite.IgniteSystemProperties.getBoolean;
import static org.apache.ignite.IgniteSystemProperties.getLong;
-import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_ALL;
-import static org.apache.ignite.cache.PartitionLossPolicy.READ_ONLY_SAFE;
-import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_ALL;
-import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
@@ -109,8 +103,8 @@ import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYS
* Future for exchanging partition maps.
*/
@SuppressWarnings({"TypeMayBeWeakened", "unchecked"})
-public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityTopologyVersion>
- implements Comparable<GridDhtPartitionsExchangeFuture>, GridDhtTopologyFuture, CachePartitionExchangeWorkerTask, IgniteDiagnosticAware {
+public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapter
+ implements Comparable<GridDhtPartitionsExchangeFuture>, CachePartitionExchangeWorkerTask, IgniteDiagnosticAware {
/** Dummy flag. */
private final boolean dummy;
@@ -194,9 +188,6 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
/** */
private CacheAffinityChangeMessage affChangeMsg;
- /** Cache groups validation results. */
- private volatile Map<Integer, CacheValidation> grpValidRes;
-
/** Skip preload flag. */
private boolean skipPreload;
@@ -648,15 +639,22 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
if (crd) {
boolean updateTop = exchId.topologyVersion().equals(grp.localStartVersion());
- if (updateTop && clientTop != null)
- top.update(exchId, clientTop.partitionMap(true), clientTop.updateCounters(false));
+ if (updateTop && clientTop != null) {
+ top.update(topologyVersion(),
+ clientTop.partitionMap(true),
+ clientTop.updateCounters(false));
+ }
}
- top.updateTopologyVersion(exchId, this, updSeq, cacheGroupStopping(grp.groupId()));
+ top.updateTopologyVersion(
+ this,
+ discoCache(),
+ updSeq,
+ cacheGroupStopping(grp.groupId()));
}
for (GridClientPartitionTopology top : cctx.exchange().clientTopologies())
- top.updateTopologyVersion(exchId, this, -1, cacheGroupStopping(top.groupId()));
+ top.updateTopologyVersion(this, discoCache(), -1, cacheGroupStopping(top.groupId()));
}
/**
@@ -676,12 +674,11 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
changeGlobalStateExceptions.put(cctx.localNodeId(), changeGlobalStateE);
}
- boolean clientOnly = cctx.affinity().onCacheChangeRequest(this, crd, exchActions);
+ assert !exchActions.clientOnlyExchange() : exchActions;
- if (clientOnly)
- return exchActions.clientCacheStarted(cctx.localNodeId()) ? ExchangeType.CLIENT : ExchangeType.NONE;
- else
- return cctx.kernalContext().clientNode() ? ExchangeType.CLIENT : ExchangeType.ALL;
+ cctx.affinity().onCacheChangeRequest(this, crd, exchActions);
+
+ return cctx.kernalContext().clientNode() ? ExchangeType.CLIENT : ExchangeType.ALL;
}
/**
@@ -761,7 +758,9 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
assert fullMap != null;
- grp.topology().update(exchId, fullMap, top.updateCounters(false));
+ grp.topology().update(topologyVersion(),
+ fullMap,
+ top.updateCounters(false));
break;
}
@@ -1191,24 +1190,15 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
discoEvt.type() == EVT_NODE_JOINED)
detectLostPartitions();
- Map<Integer, CacheValidation> m = new HashMap<>(cctx.cache().cacheGroups().size());
-
- for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
- Collection<Integer> lostParts = grp.isLocal() ?
- Collections.<Integer>emptyList() : grp.topology().lostPartitions();
-
- boolean valid = true;
+ Map<Integer, CacheValidation> m = U.newHashMap(cctx.cache().cacheGroups().size());
- if (grp.topologyValidator() != null && !grp.systemCache())
- valid = grp.topologyValidator().validate(discoEvt.topologyNodes());
-
- m.put(grp.groupId(), new CacheValidation(valid, lostParts));
- }
+ for (CacheGroupContext grp : cctx.cache().cacheGroups())
+ m.put(grp.groupId(), validateCacheGroup(grp, discoEvt.topologyNodes()));
grpValidRes = m;
}
- cctx.cache().onExchangeDone(exchId.topologyVersion(), exchActions, err, false);
+ cctx.cache().onExchangeDone(exchId.topologyVersion(), exchActions, err);
cctx.exchange().onExchangeDone(this, err);
@@ -1243,113 +1233,6 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
return dummy;
}
- /** {@inheritDoc} */
- @Nullable @Override public Throwable validateCache(
- GridCacheContext cctx,
- boolean recovery,
- boolean read,
- @Nullable Object key,
- @Nullable Collection<?> keys
- ) {
- assert isDone() : this;
-
- Throwable err = error();
-
- if (err != null)
- return err;
-
- if (!cctx.shared().kernalContext().state().active())
- return new CacheInvalidStateException(
- "Failed to perform cache operation (cluster is not activated): " + cctx.name());
-
- CacheGroupContext grp = cctx.group();
-
- PartitionLossPolicy partLossPlc = grp.config().getPartitionLossPolicy();
-
- if (grp.needsRecovery() && !recovery) {
- if (!read && (partLossPlc == READ_ONLY_SAFE || partLossPlc == READ_ONLY_ALL))
- return new IgniteCheckedException("Failed to write to cache (cache is moved to a read-only state): " +
- cctx.name());
- }
-
- if (grp.needsRecovery() || grp.topologyValidator() != null) {
- CacheValidation validation = grpValidRes.get(grp.groupId());
-
- if (validation == null)
- return null;
-
- if (!validation.valid && !read)
- return new IgniteCheckedException("Failed to perform cache operation " +
- "(cache topology is not valid): " + cctx.name());
-
- if (recovery || !grp.needsRecovery())
- return null;
-
- if (key != null) {
- int p = cctx.affinity().partition(key);
-
- CacheInvalidStateException ex = validatePartitionOperation(cctx.name(), read, key, p,
- validation.lostParts, partLossPlc);
-
- if (ex != null)
- return ex;
- }
-
- if (keys != null) {
- for (Object k : keys) {
- int p = cctx.affinity().partition(k);
-
- CacheInvalidStateException ex = validatePartitionOperation(cctx.name(), read, k, p,
- validation.lostParts, partLossPlc);
-
- if (ex != null)
- return ex;
- }
- }
- }
-
- return null;
- }
-
- /**
- * @param cacheName Cache name.
- * @param read Read flag.
- * @param key Key to check.
- * @param part Partition this key belongs to.
- * @param lostParts Collection of lost partitions.
- * @param plc Partition loss policy.
- * @return Invalid state exception if this operation is disallowed.
- */
- private CacheInvalidStateException validatePartitionOperation(
- String cacheName,
- boolean read,
- Object key,
- int part,
- Collection<Integer> lostParts,
- PartitionLossPolicy plc
- ) {
- if (lostParts.contains(part)) {
- if (!read) {
- assert plc == READ_WRITE_ALL || plc == READ_WRITE_SAFE;
-
- if (plc == READ_WRITE_SAFE) {
- return new CacheInvalidStateException("Failed to execute cache operation " +
- "(all partition owners have left the grid, partition data has been lost) [" +
- "cacheName=" + cacheName + ", part=" + part + ", key=" + key + ']');
- }
- }
- else {
- // Read.
- if (plc == READ_ONLY_SAFE || plc == READ_WRITE_SAFE)
- return new CacheInvalidStateException("Failed to execute cache operation " +
- "(all partition owners have left the grid, partition data has been lost) [" +
- "cacheName=" + cacheName + ", part=" + part + ", key=" + key + ']');
- }
- }
-
- return null;
- }
-
/**
* Cleans up resources to avoid excessive memory usage.
*/
@@ -1844,13 +1727,19 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
- if (grp != null)
- grp.topology().update(exchId, entry.getValue(), cntrMap);
+ if (grp != null) {
+ grp.topology().update(topologyVersion(),
+ entry.getValue(),
+ cntrMap);
+ }
else {
ClusterNode oldest = cctx.discovery().oldestAliveCacheServerNode(AffinityTopologyVersion.NONE);
- if (oldest != null && oldest.isLocal())
- cctx.exchange().clientTopology(grpId, this).update(exchId, entry.getValue(), cntrMap);
+ if (oldest != null && oldest.isLocal()) {
+ cctx.exchange().clientTopology(grpId, this).update(topologyVersion(),
+ entry.getValue(),
+ cntrMap);
+ }
}
}
}
@@ -2135,26 +2024,6 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
NONE
}
- /**
- * Cache validation result.
- */
- private static class CacheValidation {
- /** Topology validation result. */
- private boolean valid;
-
- /** Lost partitions on this topology version. */
- private Collection<Integer> lostParts;
-
- /**
- * @param valid Valid flag.
- * @param lostParts Lost partitions.
- */
- private CacheValidation(boolean valid, Collection<Integer> lostParts) {
- this.valid = valid;
- this.lostParts = lostParts;
- }
- }
-
/** {@inheritDoc} */
@Override public void addDiagnosticRequest(IgniteDiagnosticPrepareContext diagCtx) {
if (!isDone()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
index 527688e..71c6b65 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java
@@ -327,6 +327,13 @@ public class IgniteTxEntry implements GridPeerDeployAware, Message {
}
/**
+ * @param ctx Cache context for this tx entry.
+ */
+ public void context(GridCacheContext<?, ?> ctx) {
+ this.ctx = ctx;
+ }
+
+ /**
* @return Flag indicating if this entry is affinity mapped to the same node.
*/
public boolean locallyMapped() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 35ee011..b958a27 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -949,6 +949,17 @@ public class IgniteTxHandler {
if (nearTx != null)
res.nearEvicted(nearTx.evicted());
+ List<IgniteTxKey> writesCacheMissed = req.nearWritesCacheMissed();
+
+ if (writesCacheMissed != null) {
+ Collection<IgniteTxKey> evicted0 = res.nearEvicted();
+
+ if (evicted0 != null)
+ writesCacheMissed.addAll(evicted0);
+
+ res.nearEvicted(writesCacheMissed);
+ }
+
if (dhtTx != null)
req.txState(dhtTx.txState());
else if (nearTx != null)
@@ -1595,7 +1606,7 @@ public class IgniteTxHandler {
* @return Remote transaction.
* @throws IgniteCheckedException If failed.
*/
- @Nullable public GridNearTxRemote startNearRemoteTx(ClassLoader ldr, UUID nodeId,
+ @Nullable private GridNearTxRemote startNearRemoteTx(ClassLoader ldr, UUID nodeId,
GridDhtTxPrepareRequest req) throws IgniteCheckedException {
if (!F.isEmpty(req.nearWrites())) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index db0395f..0877305 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -276,6 +276,33 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
cctx.gridIO().addMessageListener(TOPIC_TX, new DeadlockDetectionListener());
}
+ /**
+ * @param cacheId Cache ID.
+ */
+ public void rollbackTransactionsForCache(int cacheId) {
+ rollbackTransactionsForCache(cacheId, nearIdMap);
+
+ rollbackTransactionsForCache(cacheId, threadMap);
+ }
+
+ /**
+ * @param cacheId Cache ID.
+ * @param txMap Transactions map.
+ */
+ private void rollbackTransactionsForCache(int cacheId, ConcurrentMap<?, IgniteInternalTx> txMap) {
+ for (Map.Entry<?, IgniteInternalTx> e : txMap.entrySet()) {
+ IgniteInternalTx tx = e.getValue();
+
+ for (IgniteTxEntry entry : tx.allEntries()) {
+ if (entry.cacheId() == cacheId) {
+ rollbackTx(tx);
+
+ break;
+ }
+ }
+ }
+ }
+
/** {@inheritDoc} */
@Override public void onDisconnected(IgniteFuture reconnectFut) {
txFinishSync.onDisconnected(reconnectFut);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionForCachesSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionForCachesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionForCachesSelfTest.java
index eea1c92..222e58c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionForCachesSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridProjectionForCachesSelfTest.java
@@ -26,15 +26,16 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
@@ -131,10 +132,18 @@ public class GridProjectionForCachesSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testProjectionForDefaultCache() throws Exception {
- ClusterGroup prj = ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME);
+ final ClusterGroup prj = ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME);
assertNotNull(prj);
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return prj.nodes().size() == 3;
+ }
+ }, 5000);
+
assertEquals(3, prj.nodes().size());
+
assertTrue(prj.nodes().contains(grid(0).localNode()));
assertFalse(prj.nodes().contains(grid(1).localNode()));
assertTrue(prj.nodes().contains(grid(2).localNode()));
@@ -146,9 +155,16 @@ public class GridProjectionForCachesSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testProjectionForNamedCache() throws Exception {
- ClusterGroup prj = ignite.cluster().forCacheNodes(CACHE_NAME);
+ final ClusterGroup prj = ignite.cluster().forCacheNodes(CACHE_NAME);
+
+ assertNotNull(prj);
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return prj.nodes().size() == 3;
+ }
+ }, 5000);
- assert prj != null;
assertEquals("Invalid projection: " + prj.nodes(), 3, prj.nodes().size());
assert !prj.nodes().contains(grid(0).localNode());
assert prj.nodes().contains(grid(1).localNode());
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
index 7cafa93..2e1f2f3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java
@@ -1369,28 +1369,43 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac
* @param cacheExists Cache exists flag.
* @param clientCache {@code True} if client node has client cache.
* @param clientNear {@code True} if client node has near-enabled client cache.
+ * @throws Exception If failed.
*/
private void checkCacheDiscoveryData(Ignite srv,
Ignite client,
- String cacheName,
+ final String cacheName,
boolean cacheExists,
- boolean clientCache,
- boolean clientNear) {
- GridDiscoveryManager srvDisco = ((IgniteKernal)srv).context().discovery();
+ final boolean clientCache,
+ boolean clientNear) throws Exception {
+ final GridDiscoveryManager srvDisco = ((IgniteKernal)srv).context().discovery();
GridDiscoveryManager clientDisco = ((IgniteKernal)client).context().discovery();
ClusterNode srvNode = ((IgniteKernal)srv).localNode();
- ClusterNode clientNode = ((IgniteKernal)client).localNode();
+ final ClusterNode clientNode = ((IgniteKernal)client).localNode();
assertFalse(srvDisco.cacheAffinityNode(clientNode, cacheName));
assertFalse(clientDisco.cacheAffinityNode(clientNode, cacheName));
assertEquals(cacheExists, srvDisco.cacheAffinityNode(srvNode, cacheName));
- if (clientNear)
+ if (clientNear) {
+ assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return srvDisco.cacheNearNode(clientNode, cacheName);
+ }
+ }, 5000));
+
assertTrue(srvDisco.cacheNearNode(clientNode, cacheName));
- else
+ }
+ else {
+ assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return F.eq(clientCache, srvDisco.cacheClientNode(clientNode, cacheName));
+ }
+ }, 5000));
+
assertEquals(clientCache, srvDisco.cacheClientNode(clientNode, cacheName));
+ }
assertEquals(cacheExists, clientDisco.cacheAffinityNode(srvNode, cacheName));
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
index 1a7e76a..f67e247 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheStopAndDestroySelfTest.java
@@ -30,6 +30,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.MemoryConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
@@ -50,24 +51,25 @@ import static org.apache.ignite.cache.CacheMode.PARTITIONED;
/**
* Checks stop and destroy methods behavior.
*/
+@SuppressWarnings("unchecked")
public class CacheStopAndDestroySelfTest extends GridCommonAbstractTest {
/** */
private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
- /** key-value used at test. */
- protected static String KEY_VAL = "1";
+ /** Key-value used at test. */
+ private static String KEY_VAL = "1";
- /** cache name 1. */
- protected static String CACHE_NAME_DHT = "cache";
+ /** Cache name 1. */
+ private static String CACHE_NAME_DHT = "cache";
- /** cache name 2. */
- protected static String CACHE_NAME_CLIENT = "cache_client";
+ /** Cache name 2. */
+ private static String CACHE_NAME_CLIENT = "cache_client";
- /** near cache name. */
- protected static String CACHE_NAME_NEAR = "cache_near";
+ /** Near cache name. */
+ private static String CACHE_NAME_NEAR = "cache_near";
- /** local cache name. */
- protected static String CACHE_NAME_LOC = "cache_local";
+ /** Local cache name. */
+ private static String CACHE_NAME_LOC = "cache_local";
/** Memory configuration to be used on client nodes with local caches. */
private static MemoryConfiguration memCfg;
@@ -121,12 +123,12 @@ public class CacheStopAndDestroySelfTest extends GridCommonAbstractTest {
public static AtomicInteger cnt = new AtomicInteger();
/** Node filter. */
- public static UUID nodeFilter;
+ static UUID nodeFilter;
/** {@inheritDoc} */
- @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackClosure)
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC)
throws IgniteSpiException {
- super.sendMessage(node, msg, ackClosure);
+ super.sendMessage(node, msg, ackC);
if (nodeFilter != null &&
node.id().equals(nodeFilter) &&
@@ -557,8 +559,6 @@ public class CacheStopAndDestroySelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testNearClose() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-2189");
-
startGridsMultiThreaded(gridCount());
IgniteCache<String, String> cache0 = grid(0).getOrCreateCache(getNearConfig());
@@ -596,9 +596,6 @@ public class CacheStopAndDestroySelfTest extends GridCommonAbstractTest {
U.sleep(1000);
- // Ensure near cache was NOT automatically updated.
- assert CountingTxRequestsToClientNodeTcpCommunicationSpi.cnt.get() == 0;
-
assert cache0.get(KEY_VAL).equals(KEY_VAL + 0);// Not affected.
assert cache1.get(KEY_VAL).equals(KEY_VAL + 0);// Not affected.
@@ -702,7 +699,10 @@ public class CacheStopAndDestroySelfTest extends GridCommonAbstractTest {
AffinityTopologyVersion topVer = grid(1).context().cache().context().exchange().lastTopologyFuture().get();
- grid(0).context().cache().context().exchange().affinityReadyFuture(topVer).get();
+ IgniteInternalFuture<?> fut = grid(0).context().cache().context().exchange().affinityReadyFuture(topVer);
+
+ if (fut != null)
+ fut.get();
grid(0).getOrCreateCache(getLocalConfig());
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheClearSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheClearSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheClearSelfTest.java
index 707e275..9b3777d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheClearSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheClearSelfTest.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache;
import java.util.Collections;
import java.util.Set;
-import java.util.UUID;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
@@ -28,10 +27,12 @@ import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jetbrains.annotations.Nullable;
@@ -163,13 +164,13 @@ public class GridCacheClearSelfTest extends GridCommonAbstractTest {
* @param cacheMode Cache mode.
* @param near Near cache flag.
* @param keys Keys to clear.
+ * @throws Exception If failed.
*/
- private void testClear(CacheMode cacheMode, boolean near, @Nullable Set<Integer> keys) {
+ private void testClear(CacheMode cacheMode, boolean near, @Nullable Set<Integer> keys) throws Exception {
Ignite client1 = client1();
Ignite client2 = client2();
- // TODO GG-11220 (use the same name when fixed).
- String cacheName = "cache-" + UUID.randomUUID();
+ final String cacheName = DEFAULT_CACHE_NAME;
try {
CacheConfiguration<Integer, Integer> cfg = new CacheConfiguration<>(cacheName);
@@ -184,6 +185,12 @@ public class GridCacheClearSelfTest extends GridCommonAbstractTest {
client2.createNearCache(cacheName, new NearCacheConfiguration<Integer, Integer>()) :
client2.<Integer, Integer>cache(cacheName);
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return ignite(0).cluster().forCacheNodes(cacheName).nodes().size() == 5;
+ }
+ }, 5000);
+
for (int i = 0; i < 10; i++)
cache1.put(i, i);
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheInitializationFailTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheInitializationFailTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheInitializationFailTest.java
index 1d616f8..19b0ea5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheInitializationFailTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheInitializationFailTest.java
@@ -41,6 +41,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
@@ -161,6 +162,7 @@ public class IgniteClientCacheInitializationFailTest extends GridCommonAbstractT
}
/**
+ * @param cacheName Cache name.
* @throws Exception If failed.
*/
private void checkCacheInitialization(final String cacheName) throws Exception {
@@ -170,8 +172,7 @@ public class IgniteClientCacheInitializationFailTest extends GridCommonAbstractT
checkFineCache(client, CACHE_NAME + 1);
- assertNull(client.cache(cacheName));
- assertNull(client.getOrCreateCache(cacheName));
+ assertNull(((IgniteKernal)client).context().cache().cache(cacheName));
checkFineCache(client, CACHE_NAME + 2);
}
@@ -210,7 +211,6 @@ public class IgniteClientCacheInitializationFailTest extends GridCommonAbstractT
else
cache = client.cache(cacheName);
-
cache.put(1, "1");
assertEquals("1", cache.get(1));
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheStartFailoverTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheStartFailoverTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheStartFailoverTest.java
new file mode 100644
index 0000000..f32e15f
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientCacheStartFailoverTest.java
@@ -0,0 +1,585 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheServerNotFoundException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.IgniteNodeAttributes;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtAffinityAssignmentResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionTopology;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ *
+ */
+public class IgniteClientCacheStartFailoverTest extends GridCommonAbstractTest {
+ /** */
+ protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setClientMode(client);
+
+ cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientStartCoordinatorFailsAtomic() throws Exception {
+ clientStartCoordinatorFails(ATOMIC);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientStartCoordinatorFailsTx() throws Exception {
+ clientStartCoordinatorFails(TRANSACTIONAL);
+ }
+
+ /**
+ * @param atomicityMode Cache atomicity mode.
+ * @throws Exception If failed.
+ */
+ private void clientStartCoordinatorFails(CacheAtomicityMode atomicityMode) throws Exception {
+ Ignite srv0 = startGrids(3);
+
+ final int KEYS = 500;
+
+ IgniteCache<Object, Object> cache = srv0.createCache(cacheConfiguration(DEFAULT_CACHE_NAME, atomicityMode, 1));
+
+ for (int i = 0; i < KEYS; i++)
+ cache.put(i, i);
+
+ client = true;
+
+ final Ignite c = startGrid(3);
+
+ TestRecordingCommunicationSpi.spi(srv0).blockMessages(GridDhtAffinityAssignmentResponse.class, c.name());
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ c.cache(DEFAULT_CACHE_NAME);
+
+ return null;
+ }
+ }, "start-cache");
+
+ U.sleep(1000);
+
+ assertFalse(fut.isDone());
+
+ stopGrid(0);
+
+ fut.get();
+
+ cache = c.cache(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < KEYS; i++) {
+ assertEquals(i, cache.get(i));
+
+ cache.put(i, i + 1);
+
+ assertEquals(i + 1, cache.get(i));
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientStartLastServerFailsAtomic() throws Exception {
+ clientStartLastServerFails(ATOMIC);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientStartLastServerFailsTx() throws Exception {
+ clientStartLastServerFails(TRANSACTIONAL);
+ }
+
+ /**
+ * @param atomicityMode Cache atomicity mode.
+ * @throws Exception If failed.
+ */
+ private void clientStartLastServerFails(CacheAtomicityMode atomicityMode) throws Exception {
+ startGrids(3);
+
+ CacheConfiguration<Object, Object> cfg = cacheConfiguration(DEFAULT_CACHE_NAME, atomicityMode, 1);
+
+ cfg.setNodeFilter(new TestNodeFilter(getTestIgniteInstanceName(1)));
+
+ Ignite srv1 = ignite(1);
+
+ srv1.createCache(cfg);
+
+ client = true;
+
+ final Ignite c = startGrid(3);
+
+ client = false;
+
+ TestRecordingCommunicationSpi.spi(srv1).blockMessages(GridDhtAffinityAssignmentResponse.class, c.name());
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ c.cache(DEFAULT_CACHE_NAME);
+
+ return null;
+ }
+ }, "start-cache");
+
+ U.sleep(1000);
+
+ assertFalse(fut.isDone());
+
+ stopGrid(1);
+
+ fut.get();
+
+ final IgniteCache<Object, Object> clientCache = c.cache(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < 10; i++) {
+ final int k = i;
+
+ GridTestUtils.assertThrows(log, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ clientCache.get(k);
+
+ return null;
+ }
+ }, CacheServerNotFoundException.class, null);
+
+ GridTestUtils.assertThrows(log, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ clientCache.put(k, k);
+
+ return null;
+ }
+ }, CacheServerNotFoundException.class, null);
+
+ GridTestUtils.assertThrows(log, new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ clientCache.remove(k);
+
+ return null;
+ }
+ }, CacheServerNotFoundException.class, null);
+ }
+
+ startGrid(1);
+
+ awaitPartitionMapExchange();
+
+ for (int i = 0; i < 100; i++) {
+ assertNull(clientCache.get(i));
+
+ clientCache.put(i, i);
+
+ assertEquals(i, clientCache.get(i));
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testRebalanceState() throws Exception {
+ final int SRVS = 3;
+
+ startGrids(SRVS);
+
+ List<String> cacheNames = startCaches(ignite(0), 100);
+
+ client = true;
+
+ Ignite c = startGrid(SRVS);
+
+ assertTrue(c.configuration().isClientMode());
+
+ awaitPartitionMapExchange();
+
+ client = false;
+
+ TestRecordingCommunicationSpi.spi(ignite(0)).blockMessages(new IgniteBiPredicate<ClusterNode, Message>() {
+ @Override public boolean apply(ClusterNode clusterNode, Message msg) {
+ return msg instanceof GridDhtPartitionsFullMessage &&
+ ((GridDhtPartitionsFullMessage) msg).exchangeId() == null;
+ }
+ });
+
+ startGrid(SRVS + 1);
+
+ for (String cacheName : cacheNames)
+ c.cache(cacheName);
+
+ U.sleep(1000);
+
+ for (int i = 0; i < SRVS + 1; i++) {
+ AffinityTopologyVersion topVer = new AffinityTopologyVersion(SRVS + 2);
+
+ IgniteKernal node = (IgniteKernal)ignite(i);
+
+ for (String cacheName : cacheNames) {
+ GridDhtPartitionTopology top = node.context().cache().internalCache(cacheName).context().topology();
+
+ assertEquals(topVer, top.topologyVersion());
+
+ assertFalse(top.rebalanceFinished(topVer));
+ }
+ }
+
+ TestRecordingCommunicationSpi.spi(ignite(0)).stopBlock();
+
+ for (int i = 0; i < SRVS + 1; i++) {
+ final AffinityTopologyVersion topVer = new AffinityTopologyVersion(SRVS + 2, 1);
+
+ final IgniteKernal node = (IgniteKernal)ignite(i);
+
+ for (String cacheName : cacheNames) {
+ final GridDhtPartitionTopology top = node.context().cache().internalCache(cacheName).context().topology();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return top.rebalanceFinished(topVer);
+ }
+ }, 5000);
+
+ assertTrue(top.rebalanceFinished(topVer));
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testRebalanceStateConcurrentStart() throws Exception {
+ final int SRVS1 = 3;
+ final int CLIENTS = 5;
+ final int SRVS2 = 5;
+
+ startGrids(SRVS1);
+
+ Ignite srv0 = ignite(0);
+
+ final int KEYS = 1000;
+
+ final List<String> cacheNames = startCaches(srv0, KEYS);
+
+ client = true;
+
+ final List<Ignite> clients = new ArrayList<>();
+
+ for (int i = 0; i < CLIENTS; i++)
+ clients.add(startGrid(SRVS1 + i));
+
+ client = false;
+
+ final CyclicBarrier barrier = new CyclicBarrier(clients.size() + SRVS2);
+
+ final AtomicInteger clientIdx = new AtomicInteger();
+
+ final Set<Integer> keys = new HashSet<>();
+
+ for (int i = 0; i < KEYS; i++)
+ keys.add(i);
+
+ IgniteInternalFuture<?> fut1 = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ barrier.await();
+
+ Ignite client = clients.get(clientIdx.getAndIncrement());
+
+ for (String cacheName : cacheNames)
+ client.cache(cacheName);
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < 10; i++) {
+ for (String cacheName : cacheNames) {
+ IgniteCache<Object, Object> cache = client.cache(cacheName);
+
+ Map<Object, Object> map0 = cache.getAll(keys);
+
+ assertEquals(KEYS, map0.size());
+
+ cache.put(rnd.nextInt(KEYS), i);
+ }
+ }
+
+ return null;
+ }
+ }, clients.size(), "client-cache-start");
+
+ final AtomicInteger srvIdx = new AtomicInteger(SRVS1 + CLIENTS);
+
+ IgniteInternalFuture<?> fut2 = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ barrier.await();
+
+ startGrid(srvIdx.incrementAndGet());
+
+ return null;
+ }
+ }, SRVS2, "node-start");
+
+ fut1.get();
+ fut2.get();
+
+ final AffinityTopologyVersion topVer = new AffinityTopologyVersion(SRVS1 + SRVS2 + CLIENTS, 1);
+
+ for (Ignite client : clients) {
+ for (String cacheName : cacheNames) {
+ final GridDhtPartitionTopology top =
+ ((IgniteKernal)client).context().cache().internalCache(cacheName).context().topology();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return top.rebalanceFinished(topVer);
+ }
+ }, 5000);
+
+ assertTrue(top.rebalanceFinished(topVer));
+ }
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientStartCloseServersRestart() throws Exception {
+ final int SRVS = 4;
+ final int CLIENTS = 4;
+
+ startGrids(SRVS);
+
+ final List<String> cacheNames = startCaches(ignite(0), 1000);
+
+ client = true;
+
+ final List<Ignite> clients = new ArrayList<>();
+
+ for (int i = 0; i < CLIENTS; i++)
+ clients.add(startGrid(SRVS + i));
+
+ client = false;
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ IgniteInternalFuture<?> restartFut = GridTestUtils.runAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ while (!stop.get()) {
+ int nodeIdx = rnd.nextInt(SRVS);
+
+ stopGrid(nodeIdx);
+
+ U.sleep(rnd.nextLong(200) + 1);
+
+ startGrid(nodeIdx);
+
+ U.sleep(rnd.nextLong(200) + 1);
+ }
+
+ return null;
+ }
+ }, "restart");
+
+ final AtomicInteger clientIdx = new AtomicInteger();
+
+ IgniteInternalFuture<?> clientsFut = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ Ignite client = clients.get(clientIdx.getAndIncrement());
+
+ assertTrue(client.configuration().isClientMode());
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ while (!stop.get()) {
+ for (String cacheName : cacheNames)
+ client.cache(cacheName);
+
+ for (String cacheName : cacheNames) {
+ IgniteCache<Object, Object> cache = client.cache(cacheName);
+
+ cache.put(rnd.nextInt(1000), rnd.nextInt());
+
+ cache.get(rnd.nextInt(1000));
+ }
+
+ for (String cacheName : cacheNames) {
+ IgniteCache<Object, Object> cache = client.cache(cacheName);
+
+ cache.close();
+ }
+ }
+
+ return null;
+ }
+ }, CLIENTS, "client-thread");
+
+ try {
+ U.sleep(10_000);
+
+ stop.set(true);
+
+ restartFut.get();
+ clientsFut.get();
+ }
+ finally {
+ stop.set(true);
+ }
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (Ignite client : clients) {
+ for (String cacheName : cacheNames) {
+ IgniteCache<Object, Object> cache = client.cache(cacheName);
+
+ for (int i = 0; i < 10; i++) {
+ Integer key = rnd.nextInt(1000);
+
+ cache.put(key, i);
+
+ assertEquals(i, cache.get(key));
+ }
+ }
+ }
+ }
+
+ /**
+ * @param node Node.
+ * @param keys Number of keys to put in caches.
+ * @return Cache names.
+ */
+ private List<String> startCaches(Ignite node, int keys) {
+ List<String> cacheNames = new ArrayList<>();
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < keys; i++)
+ map.put(i, i);
+
+ for (int i = 0; i < 3; i++) {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration("atomic-" + i, ATOMIC, i);
+
+ IgniteCache<Object, Object> cache = node.createCache(ccfg);
+
+ cacheNames.add(ccfg.getName());
+
+ cache.putAll(map);
+ }
+
+ for (int i = 0; i < 3; i++) {
+ CacheConfiguration<Object, Object> ccfg = cacheConfiguration("tx-" + i, TRANSACTIONAL, i);
+
+ IgniteCache<Object, Object> cache = node.createCache(ccfg);
+
+ cacheNames.add(ccfg.getName());
+
+ cache.putAll(map);
+ }
+
+ return cacheNames;
+ }
+
+ /**
+ * @param name Cache name.
+ * @param atomicityMode Cache atomicity mode.
+ * @param backups Number of backups.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration<Object, Object> cacheConfiguration(String name, CacheAtomicityMode atomicityMode, int backups) {
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>(name);
+
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicityMode(atomicityMode);
+ ccfg.setBackups(backups);
+
+ return ccfg;
+ }
+ /**
+ *
+ */
+ private static class TestNodeFilter implements IgnitePredicate<ClusterNode> {
+ /** */
+ private final String includeName;
+
+ /**
+ * @param includeName Node to include.
+ */
+ public TestNodeFilter(String includeName) {
+ this.includeName = includeName;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(ClusterNode node) {
+ return includeName.equals(node.attribute(IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
index 9176cbd..0cb0856 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicClientCacheStartSelfTest.java
@@ -17,17 +17,24 @@
package org.apache.ignite.internal.processors.cache;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.Callable;
import javax.cache.CacheException;
import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager;
+import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
@@ -35,8 +42,11 @@ import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
/**
@@ -267,6 +277,173 @@ public class IgniteDynamicClientCacheStartSelfTest extends GridCommonAbstractTes
}
/**
+ * @throws Exception If failed.
+ */
+ public void testStartMultipleClientCaches() throws Exception {
+ startMultipleClientCaches(null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStartMultipleClientCachesForGroup() throws Exception {
+ startMultipleClientCaches("testGrp");
+ }
+
+ /**
+ * @param grp Caches group name.
+ * @throws Exception If failed.
+ */
+ private void startMultipleClientCaches(@Nullable String grp) throws Exception {
+ final int SRVS = 1;
+
+ Ignite srv = startGrids(SRVS);
+
+ client = true;
+
+ Ignite client = startGrid(SRVS);
+
+ for (CacheAtomicityMode atomicityMode : values()) {
+ for (boolean batch : new boolean[]{false, true})
+ startCachesForGroup(srv, client, grp, atomicityMode, batch);
+ }
+ }
+
+ /**
+ * @param srv Server node.
+ * @param client Client node.
+ * @param grp Cache group.
+ * @param atomicityMode Cache atomicity mode.
+ * @param batch {@code True} if use {@link Ignite#getOrCreateCaches(Collection)} for cache creation.
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private void startCachesForGroup(Ignite srv,
+ Ignite client,
+ @Nullable String grp,
+ CacheAtomicityMode atomicityMode,
+ boolean batch) throws Exception {
+ log.info("Start caches [grp=" + grp + ", atomicity=" + atomicityMode + ", batch=" + batch + ']');
+
+ try {
+ srv.createCaches(cacheConfigurations(grp, atomicityMode));
+
+ Collection<IgniteCache> caches;
+
+ if (batch)
+ caches = client.getOrCreateCaches(cacheConfigurations(grp, atomicityMode));
+ else {
+ caches = new ArrayList<>();
+
+ for (CacheConfiguration ccfg : cacheConfigurations(grp, atomicityMode))
+ caches.add(client.getOrCreateCache(ccfg));
+ }
+
+ for (IgniteCache cache : caches)
+ checkCache(client, cache.getName(), false, false);
+
+ Map<Integer, Integer> map1 = new HashMap<>();
+ Map<Integer, Integer> map2 = new HashMap<>();
+
+ for (int i = 0; i < 100; i++) {
+ map1.put(i, i);
+ map2.put(i, i + 1);
+ }
+
+ for (IgniteCache<Integer, Integer> cache : caches) {
+ for (Map.Entry<Integer, Integer> e : map1.entrySet())
+ cache.put(e.getKey(), e.getValue());
+
+ checkCacheData(map1, cache.getName());
+
+ cache.putAll(map2);
+
+ checkCacheData(map2, cache.getName());
+ }
+
+ for (IgniteCache<Integer, Integer> cache : caches) {
+ cache.close();
+
+ checkNoCache(client, cache.getName());
+ }
+ }
+ finally {
+ for (CacheConfiguration ccfg : cacheConfigurations(grp, atomicityMode))
+ srv.destroyCache(ccfg.getName());
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("unchecked")
+ public void testStartNewAndClientCaches() throws Exception {
+ final int SRVS = 4;
+
+ Ignite srv = startGrids(SRVS);
+
+ srv.createCaches(cacheConfigurations(null, ATOMIC));
+
+ ccfg = null;
+
+ client = true;
+
+ Ignite client = startGrid(SRVS);
+
+ List<CacheConfiguration> cfgs = new ArrayList<>();
+
+ cfgs.addAll(cacheConfigurations(null, ATOMIC));
+ cfgs.addAll(cacheConfigurations(null, TRANSACTIONAL));
+
+ assertEquals(6, cfgs.size());
+
+ Collection<IgniteCache> caches = client.getOrCreateCaches(cfgs);
+
+ assertEquals(cfgs.size(), caches.size());
+
+ for (CacheConfiguration cfg : cfgs)
+ checkCache(client, cfg.getName(), false, false);
+
+ Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i);
+
+ for (IgniteCache<Object, Object> cache : caches) {
+ cache.putAll(map);
+
+ checkCacheData(map, cache.getName());
+ }
+
+ for (IgniteCache cache : caches) {
+ cache.close();
+
+ checkNoCache(client, cache.getName());
+ }
+ }
+
+ /**
+ * @param grp Group name.
+ * @param atomicityMode Atomicity mode.
+ * @return Cache configurations.
+ */
+ private List<CacheConfiguration> cacheConfigurations(@Nullable String grp, CacheAtomicityMode atomicityMode) {
+ List<CacheConfiguration> ccfgs = new ArrayList<>();
+
+ for (int i = 0; i < 3; i++) {
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setGroupName(grp);
+ ccfg.setName("cache-" + atomicityMode + "-" + i);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+
+ ccfgs.add(ccfg);
+ }
+
+ return ccfgs;
+ }
+
+ /**
* @param createFromCacheClient If {@code true} creates cache from cache client node.
* @throws Exception If failed.
*/
@@ -291,6 +468,8 @@ public class IgniteDynamicClientCacheStartSelfTest extends GridCommonAbstractTes
ignite0.cache(DEFAULT_CACHE_NAME).close();
+ checkNoCache(ignite0, DEFAULT_CACHE_NAME);
+
assertNotNull(ignite0.cache(DEFAULT_CACHE_NAME));
startGrid(2);
@@ -303,27 +482,37 @@ public class IgniteDynamicClientCacheStartSelfTest extends GridCommonAbstractTes
* @param cacheName Cache name
* @param srv {@code True} if server cache is expected.
* @param near {@code True} if near cache is expected.
+ * @throws Exception If failed.
*/
- private void checkCache(Ignite ignite, String cacheName, boolean srv, boolean near) {
+ private void checkCache(Ignite ignite, final String cacheName, boolean srv, boolean near) throws Exception {
GridCacheAdapter<Object, Object> cache = ((IgniteKernal)ignite).context().cache().internalCache(cacheName);
assertNotNull("No cache on node " + ignite.name(), cache);
assertEquals(near, cache.context().isNear());
- ClusterNode node = ((IgniteKernal)ignite).localNode();
+ final ClusterNode node = ((IgniteKernal)ignite).localNode();
for (Ignite ignite0 : Ignition.allGrids()) {
- GridDiscoveryManager disco = ((IgniteKernal)ignite0).context().discovery();
+ final GridDiscoveryManager disco = ((IgniteKernal)ignite0).context().discovery();
+
+ if (srv || ignite == ignite0)
+ assertTrue(disco.cacheNode(node, cacheName));
+ else {
+ assertTrue(ignite0.name(), GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return disco.cacheNode(node, cacheName);
+ }
+ }, 5000));
+ }
- assertTrue(disco.cacheNode(node, cacheName));
assertEquals(srv, disco.cacheAffinityNode(node, cacheName));
assertEquals(near, disco.cacheNearNode(node, cacheName));
if (srv)
- assertTrue(ignite0.affinity(DEFAULT_CACHE_NAME).primaryPartitions(node).length > 0);
+ assertTrue(ignite0.affinity(cacheName).primaryPartitions(node).length > 0);
else
- assertEquals(0, ignite0.affinity(DEFAULT_CACHE_NAME).primaryPartitions(node).length);
+ assertEquals(0, ignite0.affinity(cacheName).primaryPartitions(node).length);
}
assertNotNull(ignite.cache(cacheName));
@@ -332,24 +521,63 @@ public class IgniteDynamicClientCacheStartSelfTest extends GridCommonAbstractTes
/**
* @param ignite Node.
* @param cacheName Cache name.
+ * @throws Exception If failed.
*/
- private void checkNoCache(Ignite ignite, String cacheName) {
+ private void checkNoCache(Ignite ignite, final String cacheName) throws Exception {
GridCacheAdapter<Object, Object> cache = ((IgniteKernal)ignite).context().cache().internalCache(cacheName);
assertNull("Unexpected cache on node " + ignite.name(), cache);
- ClusterNode node = ((IgniteKernal)ignite).localNode();
+ final ClusterNode node = ((IgniteKernal)ignite).localNode();
for (Ignite ignite0 : Ignition.allGrids()) {
- GridDiscoveryManager disco = ((IgniteKernal)ignite0).context().discovery();
+ final GridDiscoveryManager disco = ((IgniteKernal)ignite0).context().discovery();
+
+ if (ignite0 == ignite)
+ assertFalse(ignite0.name(), disco.cacheNode(node, cacheName));
+ else {
+ assertTrue(ignite0.name(), GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return !disco.cacheNode(node, cacheName);
+ }
+ }, 5000));
+ }
- assertFalse(disco.cacheNode(node, cacheName));
assertFalse(disco.cacheAffinityNode(node, cacheName));
assertFalse(disco.cacheNearNode(node, cacheName));
}
}
/**
+ * @throws Exception If failed.
+ */
+ public void testStartClientCachesOnCoordinatorWithGroup() throws Exception {
+ startGrids(3);
+
+ List<CacheConfiguration> ccfgs = cacheConfigurations("testGrp", ATOMIC);
+
+ for (CacheConfiguration ccfg : ccfgs)
+ ccfg.setNodeFilter(new CachePredicate(F.asList(getTestIgniteInstanceName(0))));
+
+ ignite(1).createCaches(ccfgs);
+
+ ccfgs = cacheConfigurations("testGrp", ATOMIC);
+
+ for (CacheConfiguration ccfg : ccfgs)
+ ccfg.setNodeFilter(new CachePredicate(F.asList(getTestIgniteInstanceName(0))));
+
+ for (IgniteCache<Object, Object> cache : ignite(0).getOrCreateCaches(ccfgs)) {
+ cache.put(1, 1);
+
+ assertEquals(1, cache.get(1));
+
+ cache.close();
+ }
+
+ startGrid(4);
+ }
+
+ /**
*
*/
static class CachePredicate implements IgnitePredicate<ClusterNode> {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteNearClientCacheCloseTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteNearClientCacheCloseTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteNearClientCacheCloseTest.java
new file mode 100644
index 0000000..58fb512
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteNearClientCacheCloseTest.java
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ *
+ */
+public class IgniteNearClientCacheCloseTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setClientMode(client);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearCacheCloseAtomic1() throws Exception {
+ nearCacheClose(1, false, ATOMIC);
+
+ nearCacheClose(1, true, ATOMIC);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearCacheCloseAtomic2() throws Exception {
+ nearCacheClose(4, false, ATOMIC);
+
+ nearCacheClose(4, true, ATOMIC);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearCacheCloseTx1() throws Exception {
+ nearCacheClose(1, false, TRANSACTIONAL);
+
+ nearCacheClose(1, true, TRANSACTIONAL);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testNearCacheCloseTx2() throws Exception {
+ nearCacheClose(4, false, TRANSACTIONAL);
+
+ nearCacheClose(4, true, TRANSACTIONAL);
+ }
+
+ /**
+ * @param srvs Number of server nodes.
+ * @param srvNearCache {@code True} to enable near cache on server nodes.
+ * @param atomicityMode Cache atomicity mode.
+ * @throws Exception If failed.
+ */
+ private void nearCacheClose(int srvs, boolean srvNearCache, CacheAtomicityMode atomicityMode) throws Exception {
+ Ignite srv;
+
+ if (Ignition.allGrids().isEmpty()) {
+ srv = startGrids(srvs);
+
+ client = true;
+
+ startGrid(srvs);
+ }
+ else
+ srv = grid(0);
+
+ IgniteCache<Object, Object> srvCache = srv.createCache(cacheConfiguration(atomicityMode, srvNearCache));
+
+ List<Integer> keys = new ArrayList<>();
+
+ keys.add(primaryKey(srvCache));
+
+ if (srvs > 1) {
+ keys.add(backupKey(srvCache));
+ keys.add(nearKey(srvCache));
+ }
+
+ for (Integer key : keys) {
+ IgniteCache<Object, Object> clientCache =
+ ignite(srvs).createNearCache(DEFAULT_CACHE_NAME, new NearCacheConfiguration<>());
+
+ clientCache.put(key, 1);
+
+ clientCache.close();
+
+ srvCache.put(key, 2);
+
+ assertEquals(2, srvCache.get(key));
+
+ srvCache.put(key, 3);
+
+ assertEquals(3, srvCache.get(key));
+ }
+
+ srvCache.destroy();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testConcurrentUpdateAndNearCacheClose() throws Exception {
+ final int SRVS = 4;
+
+ startGrids(SRVS);
+
+ client = true;
+
+ startGrid(SRVS);
+
+ startGrid(SRVS + 1);
+
+ concurrentUpdateAndNearCacheClose(ATOMIC, SRVS + 1);
+
+ concurrentUpdateAndNearCacheClose(TRANSACTIONAL, SRVS + 1);
+ }
+
+ /**
+ * @param atomicityMode Cache atomicity mode.
+ * @param nearClient Index of client node with near cache.
+ * @throws Exception If failed.
+ */
+ private void concurrentUpdateAndNearCacheClose(CacheAtomicityMode atomicityMode,
+ final int nearClient)
+ throws Exception
+ {
+ final String cacheName = ignite(0).createCache(cacheConfiguration(atomicityMode, false)).getName();
+
+ for (int iter = 0; iter < 5; iter++) {
+ log.info("Iteration: " + iter);
+
+ IgniteCache<Object, Object> nearCache = ignite(nearClient).createNearCache(cacheName,
+ new NearCacheConfiguration<>());
+
+ final int KEYS = 1000;
+
+ for (int i = 0; i < KEYS; i++)
+ nearCache.put(i, i);
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ IgniteInternalFuture<?> updateFut = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ while (!stop.get()) {
+ int node = rnd.nextInt(nearClient);
+
+ IgniteCache<Object, Object> cache = ignite(node).cache(cacheName);
+
+ if (rnd.nextBoolean()) {
+ Map<Integer, Integer> map = new TreeMap<>();
+
+ for (int i = 0; i < 10; i++)
+ map.put(rnd.nextInt(KEYS), i);
+
+ cache.putAll(map);
+ }
+ else
+ cache.put(rnd.nextInt(KEYS), node);
+ }
+
+ return null;
+ }
+ }, 10, "update");
+
+ try {
+ U.sleep(3000);
+
+ nearCache.close();
+
+ stop.set(true);
+
+ updateFut.get();
+ }
+ finally {
+ stop.set(true);
+ }
+ }
+
+ ignite(0).destroyCache(cacheName);
+ }
+
+ /**
+ * @param atomicityMode Cache atomicity mode.
+ * @param nearCache {@code True} to enable near cache.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration cacheConfiguration(CacheAtomicityMode atomicityMode, boolean nearCache) {
+ CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
+
+ if (nearCache)
+ ccfg.setNearConfiguration(new NearCacheConfiguration());
+
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicityMode(atomicityMode);
+ ccfg.setBackups(1);
+
+ return ccfg;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4026ddcc/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractCacheTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractCacheTest.java
index f94babe..9860199 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractCacheTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTopologyValidatorAbstractCacheTest.java
@@ -19,13 +19,18 @@ package org.apache.ignite.internal.processors.cache;
import java.io.Serializable;
import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
import javax.cache.CacheException;
+import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.TopologyValidator;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.transactions.Transaction;
/**
@@ -33,49 +38,71 @@ import org.apache.ignite.transactions.Transaction;
*/
public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCacheAbstractTest implements Serializable {
/** key-value used at test. */
- protected static String KEY_VAL = "1";
+ private static String KEY_VAL = "1";
/** cache name 1. */
- protected static String CACHE_NAME_1 = "cache1";
+ static String CACHE_NAME_1 = "cache1";
/** cache name 2. */
protected static String CACHE_NAME_2 = "cache2";
+ /** */
+ private boolean client;
+
/** {@inheritDoc} */
- @Override protected int gridCount() {
+ @Override protected final int gridCount() {
return 1;
}
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
- IgniteConfiguration iCfg = super.getConfiguration(igniteInstanceName);
-
- CacheConfiguration cCfg0 = cacheConfiguration(igniteInstanceName);
-
- CacheConfiguration cCfg1 = cacheConfiguration(igniteInstanceName);
- cCfg1.setName(CACHE_NAME_1);
-
- CacheConfiguration cCfg2 = cacheConfiguration(igniteInstanceName);
- cCfg2.setName(CACHE_NAME_2);
-
- iCfg.setCacheConfiguration(cCfg0, cCfg1, cCfg2);
-
- for (CacheConfiguration cCfg : iCfg.getCacheConfiguration()) {
- if (cCfg.getName().equals(CACHE_NAME_1))
- cCfg.setTopologyValidator(new TopologyValidator() {
- @Override public boolean validate(Collection<ClusterNode> nodes) {
- return nodes.size() == 2;
- }
- });
- else if (cCfg.getName().equals(CACHE_NAME_2))
- cCfg.setTopologyValidator(new TopologyValidator() {
- @Override public boolean validate(Collection<ClusterNode> nodes) {
- return nodes.size() >= 2;
- }
- });
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ if (!client) {
+ CacheConfiguration cCfg0 = cacheConfiguration(igniteInstanceName);
+
+ CacheConfiguration cCfg1 = cacheConfiguration(igniteInstanceName);
+ cCfg1.setName(CACHE_NAME_1);
+
+ CacheConfiguration cCfg2 = cacheConfiguration(igniteInstanceName);
+ cCfg2.setName(CACHE_NAME_2);
+
+ cfg.setCacheConfiguration(cCfg0, cCfg1, cCfg2);
+
+ for (CacheConfiguration cCfg : cfg.getCacheConfiguration()) {
+ if (cCfg.getName().equals(CACHE_NAME_1))
+ cCfg.setTopologyValidator(new TopologyValidator() {
+ @Override public boolean validate(Collection<ClusterNode> nodes) {
+ return servers(nodes) == 2;
+ }
+ });
+ else if (cCfg.getName().equals(CACHE_NAME_2))
+ cCfg.setTopologyValidator(new TopologyValidator() {
+ @Override public boolean validate(Collection<ClusterNode> nodes) {
+ return servers(nodes) >= 2;
+ }
+ });
+ }
+ }
+
+ cfg.setClientMode(client);
+
+ return cfg;
+ }
+
+ /**
+ * @param nodes Nodes.
+ * @return Number of server nodes.
+ */
+ private static int servers(Collection<ClusterNode> nodes) {
+ int c = 0;
+
+ for (ClusterNode node : nodes) {
+ if (!CU.clientNode(node))
+ c++;
}
- return iCfg;
+ return c;
}
/**
@@ -83,11 +110,16 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
*
* @param cacheName cache name.
*/
- protected void putInvalid(String cacheName) {
+ void putInvalid(String cacheName) {
try {
- grid(0).cache(cacheName).put(KEY_VAL, KEY_VAL);
+ List<Ignite> nodes = nodes();
- assert false : "topology validation broken";
+ assertFalse(nodes.isEmpty());
+
+ for (Ignite node : nodes)
+ node.cache(cacheName).put(KEY_VAL, KEY_VAL);
+
+ fail("Topology validation broken");
}
catch (CacheException ex) {
assert ex.getCause() instanceof IgniteCheckedException &&
@@ -100,11 +132,17 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
*
* @param cacheName cache name.
*/
- protected void putValid(String cacheName) {
+ void putValid(String cacheName) {
try {
- grid(0).cache(cacheName).put(KEY_VAL, KEY_VAL);
+ List<Ignite> nodes = nodes();
- assert grid(0).cache(cacheName).get(KEY_VAL).equals(KEY_VAL);
+ assertFalse(nodes.isEmpty());
+
+ for (Ignite node : nodes)
+ node.cache(cacheName).put(KEY_VAL, KEY_VAL);
+
+ for (Ignite node : nodes)
+ assertEquals(KEY_VAL, node.cache(cacheName).get(KEY_VAL));
}
catch (CacheException ignored) {
assert false : "topology validation broken";
@@ -116,13 +154,13 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
*
* @param cacheName cache name.
*/
- protected void getInvalid(String cacheName) {
- try {
- assert grid(0).cache(cacheName).get(KEY_VAL).equals(KEY_VAL);
- }
- catch (CacheException ignored) {
- assert false : "topology validation broken";
- }
+ void getInvalid(String cacheName) {
+ List<Ignite> nodes = nodes();
+
+ assertFalse(nodes.isEmpty());
+
+ for (Ignite node : nodes)
+ assertEquals(KEY_VAL, node.cache(cacheName).get(KEY_VAL));
}
/**
@@ -130,11 +168,16 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
*
* @param cacheName cache name.
*/
- protected void removeInvalid(String cacheName) {
+ void removeInvalid(String cacheName) {
try {
- grid(0).cache(cacheName).remove(KEY_VAL);
+ List<Ignite> nodes = nodes();
- assert false : "topology validation broken";
+ assertFalse(nodes.isEmpty());
+
+ for (Ignite node : nodes)
+ node.cache(cacheName).remove(KEY_VAL);
+
+ fail("Topology validation broken");
}
catch (CacheException ex) {
assert ex.getCause() instanceof IgniteCheckedException &&
@@ -143,13 +186,26 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
}
/**
+ * @return Nodes.
+ */
+ private List<Ignite> nodes() {
+ if (this instanceof IgniteTopologyValidatorAbstractTxCacheTest ||
+ this instanceof IgniteTopologyValidatorAbstractTxCacheGroupsTest)
+ return Collections.singletonList(ignite(0));
+ else
+ return G.allGrids();
+ }
+
+ /**
* Commits with error.
*
* @param tx transaction.
*/
- protected void commitFailed(Transaction tx) {
+ void commitFailed(Transaction tx) {
try {
tx.commit();
+
+ fail();
}
catch (IgniteException ex) {
assert ex.getCause() instanceof IgniteCheckedException &&
@@ -163,7 +219,12 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
* @param cacheName cache name.
*/
public void remove(String cacheName) {
- assert grid(0).cache(cacheName).get(KEY_VAL) != null;
+ List<Ignite> nodes = G.allGrids();
+
+ assertFalse(nodes.isEmpty());
+
+ for (Ignite node : nodes)
+ assertNotNull(node.cache(cacheName).get(KEY_VAL));
grid(0).cache(cacheName).remove(KEY_VAL);
}
@@ -173,13 +234,14 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
*
* @param cacheName cache name.
*/
- public void assertEmpty(String cacheName) {
- assert grid(0).cache(cacheName).get(KEY_VAL) == null;
+ void assertEmpty(String cacheName) {
+ assertNull(grid(0).cache(cacheName).get(KEY_VAL));
}
- /** topology validator test. */
+ /**
+ * @throws Exception If failed.
+ */
public void testTopologyValidator() throws Exception {
-
putValid(DEFAULT_CACHE_NAME);
remove(DEFAULT_CACHE_NAME);
@@ -210,5 +272,19 @@ public abstract class IgniteTopologyValidatorAbstractCacheTest extends IgniteCac
putValid(CACHE_NAME_2);
remove(CACHE_NAME_2);
+
+ client = true;
+
+ startGrid(3);
+
+ putValid(DEFAULT_CACHE_NAME);
+ remove(DEFAULT_CACHE_NAME);
+
+ getInvalid(CACHE_NAME_1);
+ putInvalid(CACHE_NAME_1);
+ removeInvalid(CACHE_NAME_1);
+
+ putValid(CACHE_NAME_2);
+ remove(CACHE_NAME_2);
}
}