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 2015/06/10 18:27:34 UTC
[11/31] incubator-ignite git commit: ignite-471-2: huge merge from
sprint-6
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java
index 85256b4..f996568 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java
@@ -21,10 +21,11 @@ import org.apache.ignite.*;
import org.apache.ignite.cache.*;
import org.apache.ignite.cache.affinity.rendezvous.*;
import org.apache.ignite.configuration.*;
-import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import java.io.*;
import java.util.concurrent.atomic.*;
import static org.apache.ignite.cache.CacheMode.*;
@@ -50,20 +51,24 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
super.beforeTestsStarted();
- if (!clientOnly())
- grid(nearOnlyGridName).createNearCache(null, new NearCacheConfiguration());
+ if (nearEnabled())
+ grid(nearOnlyGridName).createNearCache(null, nearConfiguration());
}
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
- if (gridCnt.getAndIncrement() == 0) {
+ int cnt = gridCnt.incrementAndGet();
+
+ if ((cnt == gridCount() && isClientStartedLast()) || (cnt == 1 && !isClientStartedLast())) {
cfg.setClientMode(true);
nearOnlyGridName = gridName;
}
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true);
+
return cfg;
}
@@ -75,17 +80,14 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
cfg.setCacheStoreFactory(null);
cfg.setReadThrough(false);
cfg.setWriteThrough(false);
- cfg.setAffinity(new RendezvousAffinityFunction(false, 32));
cfg.setBackups(1);
- return cfg;
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- super.afterTestsStopped();
+ if (cfg.getCacheMode() == REPLICATED)
+ cfg.setAffinity(null);
+ else
+ cfg.setAffinity(new RendezvousAffinityFunction(false, 32));
- gridCnt.set(0);
+ return cfg;
}
/** {@inheritDoc} */
@@ -94,9 +96,11 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
}
/**
- * @return If {@code true} then uses CLIENT_ONLY mode, otherwise NEAR_ONLY.
+ * @return boolean {@code True} if client's grid must be started last, {@code false} if it must be started first.
*/
- protected abstract boolean clientOnly();
+ protected boolean isClientStartedLast() {
+ return false;
+ }
/**
* @throws Exception If failed.
@@ -110,7 +114,7 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
nearOnly.putAll(F.asMap(5, 5, 6, 6, 7, 7, 8, 8, 9, 9));
for (int key = 0; key < 10; key++) {
- for (int i = 1; i < gridCount(); i++) {
+ for (int i = 0; i < gridCount(); i++) {
if (grid(i).affinity(null).isPrimaryOrBackup(grid(i).localNode(), key))
assertEquals(key, grid(i).cache(null).localPeek(key, CachePeekMode.ONHEAP));
}
@@ -120,6 +124,24 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
assertNull(nearOnly.localPeek(key, CachePeekMode.PRIMARY, CachePeekMode.BACKUP));
}
+
+ Integer key = 1000;
+
+ nearOnly.put(key, new TestClass1(key));
+
+ if (nearEnabled())
+ assertNotNull(nearOnly.localPeek(key, CachePeekMode.ALL));
+ else
+ assertNull(nearOnly.localPeek(key, CachePeekMode.ALL));
+
+ for (int i = 0; i < gridCount(); i++) {
+ if (grid(i).affinity(null).isPrimaryOrBackup(grid(i).localNode(), key)) {
+ TestClass1 val = (TestClass1)grid(i).cache(null).localPeek(key, CachePeekMode.ONHEAP);
+
+ assertNotNull(val);
+ assertEquals(key.intValue(), val.val);
+ }
+ }
}
/**
@@ -147,6 +169,18 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
if (nearEnabled())
assertEquals(key, nearOnly.localPeek(key, CachePeekMode.ONHEAP));
}
+
+ Integer key = 2000;
+
+ dht.put(key, new TestClass2(key));
+
+ TestClass2 val = (TestClass2)nearOnly.get(key);
+
+ assertNotNull(val);
+ assertEquals(key.intValue(), val.val);
+
+ if (nearEnabled())
+ assertNotNull(nearOnly.localPeek(key, CachePeekMode.ONHEAP));
}
/**
@@ -210,4 +244,34 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst
return null;
}
+
+ /**
+ *
+ */
+ static class TestClass1 implements Serializable {
+ /** */
+ int val;
+
+ /**
+ * @param val Value.
+ */
+ public TestClass1(int val) {
+ this.val = val;
+ }
+ }
+
+ /**
+ *
+ */
+ static class TestClass2 implements Serializable {
+ /** */
+ int val;
+
+ /**
+ * @param val Value.
+ */
+ public TestClass2(int val) {
+ this.val = val;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesTcpClientDiscoveryAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesTcpClientDiscoveryAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesTcpClientDiscoveryAbstractTest.java
new file mode 100644
index 0000000..bbc9144
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesTcpClientDiscoveryAbstractTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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;
+
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+
+import static org.apache.ignite.cache.CacheMode.*;
+
+/**
+ * Tests {@link org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi} in client mode.
+ */
+@SuppressWarnings("RedundantMethodOverride")
+public abstract class GridCacheClientModesTcpClientDiscoveryAbstractTest extends GridCacheClientModesAbstractSelfTest {
+ /** {@inheritDoc} */
+ @Override protected boolean isClientStartedLast() {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(false);
+
+ return cfg;
+ }
+
+ /** */
+ public static class CaseNearReplicatedAtomic extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return REPLICATED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.ATOMIC;
+ }
+ }
+
+ /** */
+ public static class CaseNearReplicatedTransactional extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return REPLICATED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.TRANSACTIONAL;
+ }
+ }
+
+ /** */
+ public static class CaseNearPartitionedAtomic extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.ATOMIC;
+ }
+ }
+
+ /** */
+ public static class CaseNearPartitionedTransactional extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.TRANSACTIONAL;
+ }
+ }
+
+ /** */
+ public static class CaseClientReplicatedAtomic extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected NearCacheConfiguration nearConfiguration() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return REPLICATED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.ATOMIC;
+ }
+ }
+
+ /** */
+ public static class CaseClientReplicatedTransactional extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected NearCacheConfiguration nearConfiguration() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return REPLICATED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.TRANSACTIONAL;
+ }
+ }
+
+ /** */
+ public static class CaseClientPartitionedAtomic extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected NearCacheConfiguration nearConfiguration() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.ATOMIC;
+ }
+ }
+
+ /** */
+ public static class CaseClientPartitionedTransactional extends GridCacheClientModesTcpClientDiscoveryAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected NearCacheConfiguration nearConfiguration() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheMode cacheMode() {
+ return PARTITIONED;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CacheAtomicityMode atomicityMode() {
+ return CacheAtomicityMode.TRANSACTIONAL;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheMixedModeSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheMixedModeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheMixedModeSelfTest.java
index e19442f..a3c977f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheMixedModeSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheMixedModeSelfTest.java
@@ -21,6 +21,7 @@ import org.apache.ignite.*;
import org.apache.ignite.cache.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.spi.discovery.tcp.*;
import org.apache.ignite.testframework.junits.common.*;
/**
@@ -31,6 +32,8 @@ public class GridCacheMixedModeSelfTest extends GridCommonAbstractTest {
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true);
+
cfg.setCacheConfiguration(cacheConfiguration(gridName));
if (F.eq(gridName, getTestGridName(0)))
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
new file mode 100644
index 0000000..6782ff4
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
@@ -0,0 +1,1803 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.managers.communication.*;
+import org.apache.ignite.internal.processors.affinity.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.*;
+import org.apache.ignite.internal.processors.cache.distributed.near.*;
+import org.apache.ignite.internal.processors.cache.version.*;
+import org.apache.ignite.internal.util.lang.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.apache.ignite.spi.communication.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.testframework.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.apache.ignite.transactions.*;
+import org.eclipse.jetty.util.*;
+import org.jetbrains.annotations.*;
+
+import javax.cache.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.*;
+
+import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.*;
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+import static org.apache.ignite.cache.CacheMode.*;
+import static org.apache.ignite.cache.CacheRebalanceMode.*;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
+import static org.apache.ignite.transactions.TransactionConcurrency.*;
+import static org.apache.ignite.transactions.TransactionIsolation.*;
+
+/**
+ *
+ */
+public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstractTest {
+ /** */
+ protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private CacheConfiguration ccfg;
+
+ /** */
+ private boolean client;
+
+ /** */
+ private volatile CyclicBarrier updateBarrier;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder).setForceServerMode(true);
+
+ cfg.setClientMode(client);
+
+ cfg.setCommunicationSpi(new TestCommunicationSpi());
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPutAllClockMode() throws Exception {
+ atomicPut(CLOCK, true, null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPutAllPrimaryMode() throws Exception {
+ atomicPut(PRIMARY, true, null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPutAllNearEnabledClockMode() throws Exception {
+ atomicPut(CLOCK, true, new NearCacheConfiguration());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPutAllNearEnabledPrimaryMode() throws Exception {
+ atomicPut(PRIMARY, true, new NearCacheConfiguration());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPutClockMode() throws Exception {
+ atomicPut(CLOCK, false, null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPutPrimaryMode() throws Exception {
+ atomicPut(PRIMARY, false, null);
+ }
+
+ /**
+ * @param writeOrder Write order.
+ * @param putAll If {@code true} executes putAll.
+ * @param nearCfg Near cache configuration.
+ * @throws Exception If failed.
+ */
+ private void atomicPut(CacheAtomicWriteOrderMode writeOrder,
+ final boolean putAll,
+ @Nullable NearCacheConfiguration nearCfg) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(ATOMIC);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicWriteOrderMode(writeOrder);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+
+ ccfg.setNearConfiguration(nearCfg);
+
+ client = true;
+
+ ccfg.setNearConfiguration(null);
+
+ Ignite ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ final int KEYS = putAll ? 100 : 1;
+
+ for (int i = 0; i < KEYS; i++)
+ map.put(i, i);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi();
+
+ // Block messages requests for both nodes.
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite1.localNode().id());
+
+ final IgniteCache<Integer, Integer> cache = ignite2.cache(null);
+
+ assertEquals(writeOrder, cache.getConfiguration(CacheConfiguration.class).getAtomicWriteOrderMode());
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ if (putAll)
+ cache.putAll(map);
+ else
+ cache.put(0, 0);
+
+ return null;
+ }
+ });
+
+ assertFalse(putFut.isDone());
+
+ client = false;
+
+ IgniteEx ignite3 = startGrid(3);
+
+ log.info("Stop block1.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ checkData(map, null, cache, 4);
+
+ ignite3.close();
+
+ map.clear();
+
+ for (int i = 0; i < KEYS; i++)
+ map.put(i, i + 1);
+
+ // Block messages requests for single node.
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite0.localNode().id());
+
+ putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ if (putAll)
+ cache.putAll(map);
+ else
+ cache.put(0, 1);
+
+ return null;
+ }
+ });
+
+ assertFalse(putFut.isDone());
+
+ client = false;
+
+ startGrid(3);
+
+ log.info("Stop block2.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ checkData(map, null, cache, 4);
+
+ for (int i = 0; i < KEYS; i++)
+ map.put(i, i + 2);
+
+ if (putAll)
+ cache.putAll(map);
+ else
+ cache.put(0, 2);
+
+ checkData(map, null, cache, 4);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicNoRemapClockMode() throws Exception {
+ atomicNoRemap(CLOCK);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicNoRemapPrimaryMode() throws Exception {
+ atomicNoRemap(PRIMARY);
+ }
+
+ /**
+ * @param writeOrder Write order.
+ * @throws Exception If failed.
+ */
+ private void atomicNoRemap(CacheAtomicWriteOrderMode writeOrder) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(ATOMIC);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicWriteOrderMode(writeOrder);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ map.put(primaryKey(ignite0.cache(null)), 0);
+ map.put(primaryKey(ignite1.cache(null)), 1);
+ map.put(primaryKey(ignite2.cache(null)), 2);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ // Block messages requests for both nodes.
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite1.localNode().id());
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite2.localNode().id());
+
+ spi.record(GridNearAtomicUpdateRequest.class);
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ assertEquals(writeOrder, cache.getConfiguration(CacheConfiguration.class).getAtomicWriteOrderMode());
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ cache.putAll(map);
+
+ return null;
+ }
+ });
+
+ IgniteEx ignite4 = startGrid(4);
+
+ assertTrue(ignite4.configuration().isClientMode());
+
+ assertFalse(putFut.isDone());
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ spi.record(null);
+
+ checkData(map, null, cache, 5);
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(3, msgs.size());
+
+ for (Object msg : msgs)
+ assertTrue(((GridNearAtomicUpdateRequest)msg).clientRequest());
+
+ map.put(primaryKey(ignite0.cache(null)), 3);
+ map.put(primaryKey(ignite1.cache(null)), 4);
+ map.put(primaryKey(ignite2.cache(null)), 5);
+
+ cache.putAll(map);
+
+ checkData(map, null, cache, 5);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicGetAndPutClockMode() throws Exception {
+ atomicGetAndPut(CLOCK);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicGetAndPutPrimaryMode() throws Exception {
+ atomicGetAndPut(PRIMARY);
+ }
+
+ /**
+ * @param writeOrder Write order.
+ * @throws Exception If failed.
+ */
+ private void atomicGetAndPut(CacheAtomicWriteOrderMode writeOrder) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(ATOMIC);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicWriteOrderMode(writeOrder);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+
+ client = true;
+
+ ignite0.cache(null).put(0, 0);
+
+ Ignite ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ map.put(0, 1);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi();
+
+ // Block messages requests for both nodes.
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite1.localNode().id());
+
+ final IgniteCache<Integer, Integer> cache = ignite2.cache(null);
+
+ assertEquals(writeOrder, cache.getConfiguration(CacheConfiguration.class).getAtomicWriteOrderMode());
+
+ IgniteInternalFuture<Integer> putFut = GridTestUtils.runAsync(new Callable<Integer>() {
+ @Override public Integer call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ return cache.getAndPut(0, 1);
+ }
+ });
+
+ assertFalse(putFut.isDone());
+
+ client = false;
+
+ startGrid(3);
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ Integer old = putFut.get();
+
+ checkData(map, null, cache, 4);
+
+ assertEquals((Object)0, old);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxPutAll() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+
+ client = true;
+
+ Ignite ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi();
+
+ spi.blockMessages(GridNearTxPrepareRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearTxPrepareRequest.class, ignite1.localNode().id());
+
+ final IgniteCache<Integer, Integer> cache = ignite2.cache(null);
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ cache.putAll(map);
+
+ return null;
+ }
+ });
+
+ assertFalse(putFut.isDone());
+
+ client = false;
+
+ IgniteEx ignite3 = startGrid(3);
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ checkData(map, null, cache, 4);
+
+ map.clear();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i + 1);
+
+ cache.putAll(map);
+
+ checkData(map, null, cache, 4);
+ }
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTx() throws Exception {
+ pessimisticTx(null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxNearEnabled() throws Exception {
+ pessimisticTx(new NearCacheConfiguration());
+ }
+
+ /**
+ * @param nearCfg Near cache configuration.
+ * @throws Exception If failed.
+ */
+ private void pessimisticTx(NearCacheConfiguration nearCfg) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+ ccfg.setNearConfiguration(nearCfg);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+
+ client = true;
+
+ final Ignite ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi();
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+
+ spi.record(GridNearLockRequest.class);
+
+ final IgniteCache<Integer, Integer> cache = ignite2.cache(null);
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ assertFalse(putFut.isDone());
+
+ client = false;
+
+ IgniteEx ignite3 = startGrid(3);
+
+ log.info("Stop block1.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ spi.record(null);
+
+ checkData(map, null, cache, 4);
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertTrue(((GridNearLockRequest)msgs.get(0)).firstClientRequest());
+ assertTrue(((GridNearLockRequest)msgs.get(1)).firstClientRequest());
+
+ for (int i = 2; i < msgs.size(); i++)
+ assertFalse(((GridNearLockRequest)msgs.get(i)).firstClientRequest());
+
+ ignite3.close();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i + 1);
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+
+ putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ for (Map.Entry<Integer, Integer> e : map.entrySet())
+ cache.put(e.getKey(), e.getValue());
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ ignite3 = startGrid(3);
+
+ log.info("Stop block2.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ checkData(map, null, cache, 4);
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i + 2);
+
+ try (Transaction tx = ignite2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+
+ checkData(map, null, cache, 4);
+ }
+
+ /**
+ * Tests specific scenario when mapping for first locked keys does not change, but changes for second one.
+ *
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTx2() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ final Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ AffinityTopologyVersion topVer1 = new AffinityTopologyVersion(4, 0);
+
+ assertEquals(topVer1, ignite0.context().cache().internalCache(null).context().topology().topologyVersion());
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ final Integer key1 = 0;
+ final Integer key2 = 7;
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite2.localNode().id());
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.put(key1, 1);
+ cache.put(key2, 2);
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ client = false;
+
+ IgniteEx ignite4 = startGrid(4);
+
+ AffinityTopologyVersion topVer2 = new AffinityTopologyVersion(5, 0);
+
+ assertEquals(topVer2, ignite0.context().cache().internalCache(null).context().topology().topologyVersion());
+
+ GridCacheAffinityManager aff = ignite0.context().cache().internalCache(null).context().affinity();
+
+ List<ClusterNode> nodes1 = aff.nodes(key1, topVer1);
+ List<ClusterNode> nodes2 = aff.nodes(key1, topVer2);
+
+ assertEquals(nodes1, nodes2);
+
+ nodes1 = aff.nodes(key2, topVer1);
+ nodes2 = aff.nodes(key2, topVer2);
+
+ assertFalse(nodes1.get(0).equals(nodes2.get(0)));
+
+ assertFalse(putFut.isDone());
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ checkData(F.asMap(key1, 1, key2, 2), null, cache, 5);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxNearEnabledNoRemap() throws Exception {
+ pessimisticTxNoRemap(new NearCacheConfiguration());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxNoRemap() throws Exception {
+ pessimisticTxNoRemap(null);
+ }
+
+ /**
+ * @param nearCfg Near cache configuration.
+ * @throws Exception If failed.
+ */
+ private void pessimisticTxNoRemap(@Nullable NearCacheConfiguration nearCfg) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+ ccfg.setNearConfiguration(nearCfg);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ final Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i);
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite2.localNode().id());
+
+ spi.record(GridNearLockRequest.class);
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ for (Map.Entry<Integer, Integer> e : map.entrySet())
+ cache.put(e.getKey(), e.getValue());
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ IgniteEx ignite4 = startGrid(4);
+
+ assertTrue(ignite4.configuration().isClientMode());
+
+ assertFalse(putFut.isDone());
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ spi.record(null);
+
+ checkData(map, null, cache, 5);
+
+ List<Object> msgs = spi.recordedMessages();
+
+ checkClientLockMessages(msgs, map.size());
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i + 1);
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+
+ checkData(map, null, cache, 5);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLock() throws Exception {
+ lock(null);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockNearEnabled() throws Exception {
+ lock(new NearCacheConfiguration());
+ }
+
+ /**
+ * @param nearCfg Near cache configuration.
+ * @throws Exception If failed.
+ */
+ private void lock(NearCacheConfiguration nearCfg) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+ ccfg.setNearConfiguration(nearCfg);
+
+ final IgniteEx ignite0 = startGrid(0);
+ final IgniteEx ignite1 = startGrid(1);
+
+ client = true;
+
+ final Ignite ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ final List<Integer> keys = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++)
+ keys.add(i);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi();
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+
+ final IgniteCache<Integer, Integer> cache = ignite2.cache(null);
+
+ final CountDownLatch lockedLatch = new CountDownLatch(1);
+
+ final CountDownLatch unlockLatch = new CountDownLatch(1);
+
+ IgniteInternalFuture<Lock> lockFut = GridTestUtils.runAsync(new Callable<Lock>() {
+ @Override public Lock call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ Lock lock = cache.lockAll(keys);
+
+ lock.lock();
+
+ log.info("Locked");
+
+ lockedLatch.countDown();
+
+ unlockLatch.await();
+
+ lock.unlock();
+
+ return lock;
+ }
+ });
+
+ client = false;
+
+ IgniteEx ignite3 = startGrid(3);
+
+ log.info("Stop block.");
+
+ assertEquals(1, lockedLatch.getCount());
+
+ spi.stopBlock();
+
+ assertTrue(lockedLatch.await(3000, TimeUnit.MILLISECONDS));
+
+ IgniteCache<Integer, Integer> cache0 = ignite0.cache(null);
+
+ for (Integer key : keys) {
+ Lock lock = cache0.lock(key);
+
+ assertFalse(lock.tryLock());
+ }
+
+ unlockLatch.countDown();
+
+ lockFut.get();
+
+ boolean wait = GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return unlocked(ignite0) && unlocked(ignite1);
+ }
+
+ private boolean unlocked(Ignite ignite) {
+ IgniteCache<Integer, Integer> cache = ignite.cache(null);
+
+ for (Integer key : keys) {
+ if (cache.isLocalLocked(key, false)) {
+ log.info("Key is locked [key=" + key + ", node=" + ignite.name() + ']');
+
+ return false;
+ }
+ }
+
+ return true;
+ }
+ }, 10_000);
+
+ assertTrue(wait);
+
+ for (Integer key : keys) {
+ Lock lock = cache0.lock(key);
+
+ assertTrue("Failed to lock: " + key, lock.tryLock());
+
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxMessageClientFirstFlag() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ spi.record(GridNearLockRequest.class);
+
+ IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.put(1, 1);
+ cache.put(2, 2);
+ cache.put(3, 3);
+
+ tx.commit();
+ }
+
+ checkClientLockMessages(spi.recordedMessages(), 3);
+
+ Map<Integer, Integer> map = new HashMap<>();
+
+ map.put(4, 4);
+ map.put(5, 5);
+ map.put(6, 6);
+ map.put(7, 7);
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+
+ checkClientLockMessages(spi.recordedMessages(), 4);
+
+ spi.record(null);
+
+ TestCommunicationSpi spi0 = (TestCommunicationSpi)ignite0.configuration().getCommunicationSpi();
+
+ spi0.record(GridNearLockRequest.class);
+
+ List<Integer> keys = primaryKeys(ignite1.cache(null), 3, 0);
+
+ IgniteCache<Integer, Integer> cache0 = ignite0.cache(null);
+
+ try (Transaction tx = ignite0.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache0.put(keys.get(0), 0);
+ cache0.put(keys.get(1), 1);
+ cache0.put(keys.get(2), 2);
+
+ tx.commit();
+ }
+
+ List<Object> msgs = spi0.recordedMessages();
+
+ assertEquals(3, msgs.size());
+
+ for (Object msg : msgs)
+ assertFalse(((GridNearLockRequest)msg).firstClientRequest());
+ }
+
+ /**
+ * @param msgs Messages.
+ * @param expCnt Expected number of messages.
+ */
+ private void checkClientLockMessages(List<Object> msgs, int expCnt) {
+ assertEquals(expCnt, msgs.size());
+
+ assertTrue(((GridNearLockRequest)msgs.get(0)).firstClientRequest());
+
+ for (int i = 1; i < msgs.size(); i++)
+ assertFalse(((GridNearLockRequest)msgs.get(i)).firstClientRequest());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testOptimisticTxMessageClientFirstFlag() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ List<Integer> keys0 = primaryKeys(ignite0.cache(null), 2, 0);
+ List<Integer> keys1 = primaryKeys(ignite1.cache(null), 2, 0);
+ List<Integer> keys2 = primaryKeys(ignite2.cache(null), 2, 0);
+
+ LinkedHashMap<Integer, Integer> map = new LinkedHashMap<>();
+
+ map.put(keys0.get(0), 1);
+ map.put(keys1.get(0), 2);
+ map.put(keys2.get(0), 3);
+ map.put(keys0.get(1), 4);
+ map.put(keys1.get(1), 5);
+ map.put(keys2.get(1), 6);
+
+ spi.record(GridNearTxPrepareRequest.class);
+
+ try (Transaction tx = ignite3.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
+ for (Map.Entry<Integer, Integer> e : map.entrySet())
+ cache.put(e.getKey(), e.getValue());
+
+ tx.commit();
+ }
+
+ checkClientPrepareMessages(spi.recordedMessages(), 6);
+
+ checkData(map, null, cache, 4);
+
+ cache.putAll(map);
+
+ checkClientPrepareMessages(spi.recordedMessages(), 6);
+
+ spi.record(null);
+
+ checkData(map, null, cache, 4);
+
+ IgniteCache<Integer, Integer> cache0 = ignite0.cache(null);
+
+ TestCommunicationSpi spi0 = (TestCommunicationSpi)ignite0.configuration().getCommunicationSpi();
+
+ spi0.record(GridNearTxPrepareRequest.class);
+
+ cache0.putAll(map);
+
+ spi0.record(null);
+
+ List<Object> msgs = spi0.recordedMessages();
+
+ assertEquals(4, msgs.size());
+
+ for (Object msg : msgs)
+ assertFalse(((GridNearTxPrepareRequest)msg).firstClientRequest());
+
+ checkData(map, null, cache, 4);
+ }
+
+ /**
+ * @param msgs Messages.
+ * @param expCnt Expected number of messages.
+ */
+ private void checkClientPrepareMessages(List<Object> msgs, int expCnt) {
+ assertEquals(expCnt, msgs.size());
+
+ assertTrue(((GridNearTxPrepareRequest)msgs.get(0)).firstClientRequest());
+
+ for (int i = 1; i < msgs.size(); i++)
+ assertFalse(((GridNearTxPrepareRequest) msgs.get(i)).firstClientRequest());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockRemoveAfterClientFailed() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+
+ client = true;
+
+ Ignite ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ IgniteCache<Integer, Integer> cache2 = ignite2.cache(null);
+
+ final Integer key = 0;
+
+ Lock lock2 = cache2.lock(key);
+
+ lock2.lock();
+
+ ignite2.close();
+
+ IgniteCache<Integer, Integer> cache0 = ignite0.cache(null);
+
+ assertFalse(cache0.isLocalLocked(key, false));
+
+ IgniteCache<Integer, Integer> cache1 = ignite1.cache(null);
+
+ assertFalse(cache1.isLocalLocked(key, false));
+
+ Lock lock1 = cache1.lock(0);
+
+ assertTrue(lock1.tryLock(5000, TimeUnit.MILLISECONDS));
+
+ lock1.unlock();
+
+ ignite2 = startGrid(2);
+
+ assertTrue(ignite2.configuration().isClientMode());
+
+ cache2 = ignite2.cache(null);
+
+ lock2 = cache2.lock(0);
+
+ assertTrue(lock2.tryLock(5000, TimeUnit.MILLISECONDS));
+
+ lock2.unlock();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockFromClientBlocksExchange() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ startGrid(0);
+ startGrid(1);
+
+ client = true;
+
+ Ignite ignite2 = startGrid(2);
+
+ IgniteCache<Integer, Integer> cache = ignite2.cache(null);
+
+ Lock lock = cache.lock(0);
+
+ lock.lock();
+
+ IgniteInternalFuture<?> startFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ client = false;
+
+ startGrid(3);
+
+ return null;
+ }
+ });
+
+ U.sleep(2000);
+
+ assertFalse(startFut.isDone());
+
+ AffinityTopologyVersion ver = new AffinityTopologyVersion(4);
+
+ List<IgniteInternalFuture<?>> futs = new ArrayList<>();
+
+ U.sleep(2000);
+
+ for (int i = 0; i < 3; i++) {
+ Ignite ignite = ignite(i);
+
+ IgniteInternalFuture<?> fut =
+ ((IgniteKernal)ignite).context().cache().context().exchange().affinityReadyFuture(ver);
+
+ assertNotNull(fut);
+
+ assertFalse(fut.isDone());
+
+ futs.add(fut);
+ }
+
+ lock.unlock();
+
+ for (IgniteInternalFuture<?> fut : futs)
+ fut.get(10_000);
+
+ startFut.get(10_000);
+ }
+
+ /**
+ * @param map Expected data.
+ * @param keys Expected keys (if expected data is not specified).
+ * @param clientCache Client cache.
+ * @param expNodes Expected nodes number.
+ * @throws Exception If failed.
+ */
+ private void checkData(final Map<Integer, Integer> map,
+ final Set<Integer> keys,
+ IgniteCache<?, ?> clientCache,
+ final int expNodes)
+ throws Exception
+ {
+ final List<Ignite> nodes = G.allGrids();
+
+ final Affinity<Integer> aff = nodes.get(0).affinity(null);
+
+ assertEquals(expNodes, nodes.size());
+
+ boolean hasNearCache = clientCache.getConfiguration(CacheConfiguration.class).getNearConfiguration() != null;
+
+ final Ignite nearCacheNode = hasNearCache ? clientCache.unwrap(Ignite.class) : null;
+
+ boolean wait = GridTestUtils.waitForCondition(new PA() {
+ @Override public boolean apply() {
+ try {
+ Set<Integer> keys0 = map != null ? map.keySet() : keys;
+
+ assertNotNull(keys0);
+
+ for (Integer key : keys0) {
+ GridCacheVersion ver = null;
+ Object val = null;
+
+ for (Ignite node : nodes) {
+ IgniteCache<Integer, Integer> cache = node.cache(null);
+
+ boolean affNode = aff.isPrimaryOrBackup(node.cluster().localNode(), key);
+
+ Object val0 = cache.localPeek(key);
+
+ if (affNode || node == nearCacheNode) {
+ if (map != null)
+ assertEquals("Unexpected value for " + node.name(), map.get(key), val0);
+ else
+ assertNotNull("Unexpected value for " + node.name(), val0);
+
+ GridCacheAdapter cache0 = ((IgniteKernal)node).internalCache(null);
+
+ if (affNode && cache0.isNear())
+ cache0 = ((GridNearCacheAdapter)cache0).dht();
+
+ GridCacheEntryEx entry = cache0.peekEx(key);
+
+ assertNotNull("No entry [node=" + node.name() + ", key=" + key + ']', entry);
+
+ GridCacheVersion ver0 = entry instanceof GridNearCacheEntry ?
+ ((GridNearCacheEntry)entry).dhtVersion() : entry.version();
+
+ assertNotNull("Null version [node=" + node.name() + ", key=" + key + ']', ver0);
+
+ if (ver == null) {
+ ver = ver0;
+ val = val0;
+ }
+ else {
+ assertEquals("Version check failed [node=" + node.name() +
+ ", key=" + key +
+ ", affNode=" + affNode +
+ ", primary=" + aff.isPrimary(node.cluster().localNode(), key) + ']',
+ ver0,
+ ver);
+
+ assertEquals("Value check failed [node=" + node.name() +
+ ", key=" + key +
+ ", affNode=" + affNode +
+ ", primary=" + aff.isPrimary(node.cluster().localNode(), key) + ']',
+ val0,
+ val);
+ }
+ }
+ else
+ assertNull("Unexpected non-null value for " + node.name(), val0);
+ }
+ }
+ }
+ catch (AssertionError e) {
+ log.info("Check failed, will retry: " + e);
+
+ return false;
+ }
+ catch (Exception e) {
+ fail("Unexpected exception: " + e);
+ }
+
+ return true;
+ }
+ }, 10_000);
+
+ assertTrue("Data check failed.", wait);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicPrimaryPutAllMultinode() throws Exception {
+ multinode(PRIMARY, TestType.PUT_ALL);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicClockPutAllMultinode() throws Exception {
+ multinode(CLOCK, TestType.PUT_ALL);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testOptimisticTxPutAllMultinode() throws Exception {
+ multinode(null, TestType.OPTIMISTIC_TX);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxPutAllMultinode() throws Exception {
+ multinode(null, TestType.PESSIMISTIC_TX);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testLockAllMultinode() throws Exception {
+ multinode(null, TestType.LOCK);
+ }
+
+ /**
+ * @param atomicWriteOrder Write order if test atomic cache.
+ * @param testType Test type.
+ * @throws Exception If failed.
+ */
+ private void multinode(final CacheAtomicWriteOrderMode atomicWriteOrder, final TestType testType)
+ throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(atomicWriteOrder != null ? ATOMIC : TRANSACTIONAL);
+ ccfg.setAtomicWriteOrderMode(atomicWriteOrder);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ final int SRV_CNT = 4;
+
+ for (int i = 0; i < SRV_CNT; i++)
+ startGrid(i);
+
+ final int CLIENT_CNT = 4;
+
+ final List<Ignite> clients = new ArrayList<>();
+
+ client = true;
+
+ for (int i = 0; i < CLIENT_CNT; i++) {
+ Ignite ignite = startGrid(SRV_CNT + i);
+
+ assertTrue(ignite.configuration().isClientMode());
+
+ clients.add(ignite);
+ }
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ final AtomicInteger threadIdx = new AtomicInteger(0);
+
+ final int THREADS = CLIENT_CNT * 3;
+
+ final ConcurrentHashSet<Integer> putKeys = new ConcurrentHashSet<>();
+
+ IgniteInternalFuture<?> fut;
+
+ try {
+ fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ int clientIdx = threadIdx.getAndIncrement() % CLIENT_CNT;
+
+ Ignite ignite = clients.get(clientIdx);
+
+ assertTrue(ignite.configuration().isClientMode());
+
+ Thread.currentThread().setName("update-thread-" + ignite.name());
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(null);
+
+ boolean useTx = testType == TestType.OPTIMISTIC_TX || testType == TestType.PESSIMISTIC_TX;
+
+ if (useTx || testType == TestType.LOCK) {
+ assertEquals(TRANSACTIONAL,
+ cache.getConfiguration(CacheConfiguration.class).getAtomicityMode());
+ }
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ int cntr = 0;
+
+ while (!stop.get()) {
+ TreeMap<Integer, Integer> map = new TreeMap<>();
+
+ for (int i = 0; i < 100; i++) {
+ Integer key = rnd.nextInt(0, 1000);
+
+ map.put(key, rnd.nextInt());
+ }
+
+ try {
+ if (testType == TestType.LOCK) {
+ Lock lock = cache.lockAll(map.keySet());
+
+ lock.lock();
+
+ lock.unlock();
+ }
+ else {
+ if (useTx) {
+ IgniteTransactions txs = ignite.transactions();
+
+ TransactionConcurrency concurrency =
+ testType == TestType.PESSIMISTIC_TX ? PESSIMISTIC : OPTIMISTIC;
+
+ try (Transaction tx = txs.txStart(concurrency, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+ }
+ else
+ cache.putAll(map);
+
+ putKeys.addAll(map.keySet());
+ }
+ }
+ catch (CacheException | IgniteException e) {
+ log.info("Operation failed, ignore: " + e);
+ }
+
+ if (++cntr % 100 == 0)
+ log.info("Iteration: " + cntr);
+
+ if (updateBarrier != null)
+ updateBarrier.await();
+ }
+
+ return null;
+ }
+ }, THREADS, "update-thread");
+
+ long stopTime = System.currentTimeMillis() + 60_000;
+
+ while (System.currentTimeMillis() < stopTime) {
+ boolean restartClient = ThreadLocalRandom.current().nextBoolean();
+
+ Integer idx = null;
+
+ if (restartClient) {
+ log.info("Start client node.");
+
+ client = true;
+
+ IgniteEx ignite = startGrid(SRV_CNT + CLIENT_CNT);
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(null);
+
+ assertNotNull(cache);
+ }
+ else {
+ idx = ThreadLocalRandom.current().nextInt(0, SRV_CNT);
+
+ log.info("Stop server node: " + idx);
+
+ stopGrid(idx);
+ }
+
+ updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() {
+ @Override public void run() {
+ updateBarrier = null;
+ }
+ });
+
+ try {
+ updateBarrier.await(30_000, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e) {
+ log.error("Failed to wait for update.");
+
+ for (Ignite ignite : G.allGrids())
+ dumpCacheDebugInfo(ignite);
+
+ U.dumpThreads(log);
+
+ CyclicBarrier barrier0 = updateBarrier;
+
+ if (barrier0 != null)
+ barrier0.reset();
+
+ fail("Failed to wait for update.");
+ }
+
+ U.sleep(500);
+
+ if (restartClient) {
+ log.info("Stop client node.");
+
+ stopGrid(SRV_CNT + CLIENT_CNT);
+ }
+ else {
+ log.info("Start server node: " + idx);
+
+ client = false;
+
+ startGrid(idx);
+ }
+
+ updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() {
+ @Override public void run() {
+ updateBarrier = null;
+ }
+ });
+
+ try {
+ updateBarrier.await(30_000, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e) {
+ log.error("Failed to wait for update.");
+
+ for (Ignite ignite : G.allGrids())
+ dumpCacheDebugInfo(ignite);
+
+ U.dumpThreads(log);
+
+ CyclicBarrier barrier0 = updateBarrier;
+
+ if (barrier0 != null)
+ barrier0.reset();
+
+ fail("Failed to wait for update.");
+ }
+
+ U.sleep(500);
+ }
+ }
+ finally {
+ stop.set(true);
+ }
+
+ fut.get(30_000);
+
+ if (testType != TestType.LOCK)
+ checkData(null, putKeys, grid(SRV_CNT).cache(null), SRV_CNT + CLIENT_CNT);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testServersLeaveOnStart() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(ATOMIC);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ Ignite ignite0 = startGrid(0);
+
+ client = true;
+
+ final AtomicInteger nodeIdx = new AtomicInteger(2);
+
+ final int CLIENTS = 10;
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ int idx = nodeIdx.getAndIncrement();
+
+ startGrid(idx);
+
+ return null;
+ }
+ }, CLIENTS, "start-client");
+
+ ignite0.close();
+
+ fut.get();
+
+ for (int i = 0; i < CLIENTS; i++) {
+ Ignite ignite = grid(i + 2);
+
+ assertEquals(CLIENTS, ignite.cluster().nodes().size());
+ }
+
+ client = false;
+
+ startGrid(0);
+ startGrid(1);
+
+ awaitPartitionMapExchange();
+
+ for (int i = 0; i < CLIENTS; i++) {
+ Ignite ignite = grid(i + 2);
+
+ IgniteCache<Integer, Integer> cache = ignite.cache(null);
+
+ cache.put(i, i);
+
+ assertEquals((Object)i, cache.get(i));
+ }
+ }
+
+ /**
+ *
+ */
+ private static class TestCommunicationSpi extends TcpCommunicationSpi {
+ /** */
+ @LoggerResource
+ private IgniteLogger log;
+
+ /** */
+ private List<T2<ClusterNode, GridIoMessage>> blockedMsgs = new ArrayList<>();
+
+ /** */
+ private Map<Class<?>, Set<UUID>> blockCls = new HashMap<>();
+
+ /** */
+ private Class<?> recordCls;
+
+ /** */
+ private List<Object> recordedMsgs = new ArrayList<>();
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException {
+ if (msg instanceof GridIoMessage) {
+ Object msg0 = ((GridIoMessage)msg).message();
+
+ synchronized (this) {
+ if (recordCls != null && msg0.getClass().equals(recordCls))
+ recordedMsgs.add(msg0);
+
+ Set<UUID> blockNodes = blockCls.get(msg0.getClass());
+
+ if (F.contains(blockNodes, node.id())) {
+ log.info("Block message [node=" + node.attribute(IgniteNodeAttributes.ATTR_GRID_NAME) +
+ ", msg=" + msg0 + ']');
+
+ blockedMsgs.add(new T2<>(node, (GridIoMessage)msg));
+
+ return;
+ }
+ }
+ }
+
+ super.sendMessage(node, msg);
+ }
+
+ /**
+ * @param recordCls Message class to record.
+ */
+ void record(@Nullable Class<?> recordCls) {
+ synchronized (this) {
+ this.recordCls = recordCls;
+ }
+ }
+
+ /**
+ * @return Recorded messages.
+ */
+ List<Object> recordedMessages() {
+ synchronized (this) {
+ List<Object> msgs = recordedMsgs;
+
+ recordedMsgs = new ArrayList<>();
+
+ return msgs;
+ }
+ }
+
+ /**
+ * @param cls Message class.
+ * @param nodeId Node ID.
+ */
+ void blockMessages(Class<?> cls, UUID nodeId) {
+ synchronized (this) {
+ Set<UUID> set = blockCls.get(cls);
+
+ if (set == null) {
+ set = new HashSet<>();
+
+ blockCls.put(cls, set);
+ }
+
+ set.add(nodeId);
+ }
+ }
+
+ /**
+ *
+ */
+ void stopBlock() {
+ synchronized (this) {
+ blockCls.clear();
+
+ for (T2<ClusterNode, GridIoMessage> msg : blockedMsgs) {
+ ClusterNode node = msg.get1();
+
+ log.info("Send blocked message: [node=" + node.attribute(IgniteNodeAttributes.ATTR_GRID_NAME) +
+ ", msg=" + msg.get2().message() + ']');
+
+ super.sendMessage(msg.get1(), msg.get2());
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ enum TestType {
+ /** */
+ PUT_ALL,
+
+ /** */
+ OPTIMISTIC_TX,
+
+ /** */
+ PESSIMISTIC_TX,
+
+ /** */
+ LOCK
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeConcurrentStart.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeConcurrentStart.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeConcurrentStart.java
new file mode 100644
index 0000000..bd74ece
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeConcurrentStart.java
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.cache.CacheRebalanceMode.*;
+
+/**
+ *
+ */
+public class IgniteCacheClientNodeConcurrentStart extends GridCommonAbstractTest {
+ /** */
+ protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final int NODES_CNT = 5;
+
+ /** */
+ private Set<Integer> clientNodes;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ assertNotNull(clientNodes);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ boolean client = false;
+
+ for (Integer clientIdx : clientNodes) {
+ if (getTestGridName(clientIdx).equals(gridName)) {
+ client = true;
+
+ break;
+ }
+ }
+
+ cfg.setClientMode(client);
+
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setBackups(0);
+ ccfg.setRebalanceMode(SYNC);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testConcurrentStart() throws Exception {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < 3; i++) {
+ try {
+ clientNodes = new HashSet<>();
+
+ while (clientNodes.size() < 2)
+ clientNodes.add(rnd.nextInt(0, NODES_CNT));
+
+ clientNodes.add(NODES_CNT - 1);
+
+ log.info("Test iteration [iter=" + i + ", clients=" + clientNodes + ']');
+
+ startGridsMultiThreaded(NODES_CNT, true);
+
+ for (int node : clientNodes) {
+ Ignite ignite = grid(node);
+
+ assertTrue(ignite.configuration().isClientMode());
+ }
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+ }
+}