ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yzhda...@apache.org
Subject [36/50] [abbrv] incubator-ignite git commit: # ignite-23
Date Wed, 27 May 2015 16:13:55 GMT
# ignite-23


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/32cb360f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/32cb360f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/32cb360f

Branch: refs/heads/ignite-943
Commit: 32cb360f736cdc84af8d461ad4c49cc2806cf996
Parents: 6432ec0
Author: sboikov <sboikov@gridgain.com>
Authored: Tue May 26 15:49:12 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Tue May 26 17:46:53 2015 +0300

----------------------------------------------------------------------
 .../dht/atomic/GridDhtAtomicCache.java          |   2 +-
 .../GridDhtPartitionsExchangeFuture.java        |   2 -
 ...niteCacheClientNodeChangingTopologyTest.java | 431 ++++++++++++++++---
 3 files changed, 380 insertions(+), 55 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32cb360f/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 9135f0a..69f5501 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
@@ -1062,7 +1062,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K,
V> {
 
                         if (ver == null) {
                             // Assign next version for update inside entries lock.
-                            ver = ctx.versions().next(req.topologyVersion());
+                            ver = ctx.versions().next(topology().topologyVersion());
 
                             if (hasNear)
                                 res.nearVersion(ver);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32cb360f/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 e04432f..1c1ebd5 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
@@ -514,8 +514,6 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
                         if (exchId.isLeft())
                             cctx.mvcc().removeExplicitNodeLocks(exchId.nodeId(), exchId.topologyVersion());
 
-                        cctx.mvcc().finishLocks(exchId.topologyVersion()).get();
-
                         onDone(exchId.topologyVersion());
 
                         skipPreload = true;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32cb360f/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
index b067797..45fa275 100644
--- 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
@@ -211,7 +211,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         putFut.get();
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         ignite3.close();
 
@@ -248,7 +248,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         putFut.get();
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         for (int i = 0; i < KEYS; i++)
             map.put(i, i + 2);
@@ -258,7 +258,106 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
         else
             cache.put(0, 2);
 
-        checkData(map, cache, 4);
+        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);
     }
 
     /**
@@ -334,7 +433,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         Integer old = putFut.get();
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         assertEquals((Object)0, old);
     }
@@ -394,7 +493,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         putFut.get();
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         map.clear();
 
@@ -403,7 +502,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         cache.putAll(map);
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
     }
     /**
      * @throws Exception If failed.
@@ -452,6 +551,8 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
         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>()
{
@@ -480,7 +581,17 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         putFut.get();
 
-        checkData(map, cache, 4);
+        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();
 
@@ -513,7 +624,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         putFut.get();
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         for (int i = 0; i < 100; i++)
             map.put(i, i + 2);
@@ -524,7 +635,192 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
             tx.commit();
         }
 
-        checkData(map, cache, 4);
+        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);
     }
 
     /**
@@ -775,7 +1071,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         checkClientPrepareMessages(spi.recordedMessages(), 6);
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         cache.putAll(map);
 
@@ -783,7 +1079,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         spi.record(null);
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
 
         IgniteCache<Integer, Integer> cache0 = ignite0.cache(null);
 
@@ -802,7 +1098,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
         for (Object msg : msgs)
             assertFalse(((GridNearTxPrepareRequest)msg).firstClientRequest());
 
-        checkData(map, cache, 4);
+        checkData(map, null, cache, 4);
     }
 
     /**
@@ -944,11 +1240,15 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
     /**
      * @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, IgniteCache<?, ?>
clientCache, final int expNodes)
+    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();
@@ -964,18 +1264,26 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
         boolean wait = GridTestUtils.waitForCondition(new PA() {
             @Override public boolean apply() {
                 try {
-                    for (Map.Entry<Integer, Integer> e : map.entrySet()) {
-                        Integer key = e.getKey();
+                    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) {
-                                assertEquals("Unexpected value for " + node.name(), e.getValue(),
cache.localPeek(key));
+                                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);
 
@@ -991,13 +1299,28 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
                                 assertNotNull("Null version [node=" + node.name() + ", key="
+ key + ']', ver0);
 
-                                if (ver == null)
+                                if (ver == null) {
                                     ver = ver0;
-                                else
-                                    assertEquals(ver0, ver);
+                                    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(),
cache.localPeek(key));
+                                assertNull("Unexpected non-null value for " + node.name(),
val0);
                         }
                     }
                 }
@@ -1080,10 +1403,6 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
             clients.add(ignite);
         }
 
-        client = false;
-
-        List<IgniteInternalFuture<?>> futs = new ArrayList<>();
-
         final AtomicBoolean stop = new AtomicBoolean();
 
         final AtomicInteger threadIdx = new AtomicInteger(0);
@@ -1092,8 +1411,10 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
         final ConcurrentHashSet<Integer> putKeys = new ConcurrentHashSet<>();
 
+        IgniteInternalFuture<?> fut;
+
         try {
-            GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
+            fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
                 @Override public Object call() throws Exception {
                     int clientIdx = threadIdx.getAndIncrement() % CLIENT_CNT;
 
@@ -1122,7 +1443,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
                         for (int i = 0; i < 100; i++) {
                             Integer key = rnd.nextInt(0, 1000);
 
-                            map.put(key, key);
+                            map.put(key, rnd.nextInt());
                         }
 
                         try {
@@ -1157,28 +1478,31 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
                 }
             }, THREADS, "update-thread");
 
-            for (final Ignite ignite : clients) {
+            long stopTime = System.currentTimeMillis() + 60_000;
 
-                futs.add(GridTestUtils.runAsync(new Callable<Object>() {
-                    @Override public Object call() throws Exception {
-                        Thread.currentThread().setName("update-" + ignite.name());
+            while (System.currentTimeMillis() < stopTime) {
+                boolean restartClient = ThreadLocalRandom.current().nextBoolean();
 
-                        log.info("Start updates from node: " + ignite.name());
+                Integer idx = null;
 
+                if (restartClient) {
+                    log.info("Start client node.");
 
-                        return null;
-                    }
-                }));
-            }
+                    client = true;
 
-            long stopTime = System.currentTimeMillis() + 60_000;
+                    IgniteEx ignite = startGrid(SRV_CNT + CLIENT_CNT);
 
-            while (System.currentTimeMillis() < stopTime) {
-                int idx = ThreadLocalRandom.current().nextInt(0, SRV_CNT);
+                    IgniteCache<Integer, Integer> cache = ignite.cache(null);
 
-                log.info("Stop node: " + idx);
+                    assertNotNull(cache);
+                }
+                else {
+                    idx = ThreadLocalRandom.current().nextInt(0, SRV_CNT);
+
+                    log.info("Stop server node: " + idx);
 
-                stopGrid(idx);
+                    stopGrid(idx);
+                }
 
                 updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() {
                     @Override public void run() {
@@ -1190,7 +1514,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
                     updateBarrier.await(30_000, TimeUnit.MILLISECONDS);
                 }
                 catch (TimeoutException e) {
-                    log.info("Failed to wait for update.");
+                    log.error("Failed to wait for update.");
 
                     U.dumpThreads(log);
 
@@ -1204,9 +1528,18 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
 
                 U.sleep(500);
 
-                log.info("Start node: " + idx);
+                if (restartClient) {
+                    log.info("Stop client node.");
 
-                startGrid(idx);
+                    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() {
@@ -1218,7 +1551,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
                     updateBarrier.await(30_000, TimeUnit.MILLISECONDS);
                 }
                 catch (TimeoutException e) {
-                    log.info("Failed to wait for update.");
+                    log.error("Failed to wait for update.");
 
                     U.dumpThreads(log);
 
@@ -1237,15 +1570,9 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
             stop.set(true);
         }
 
-        for (IgniteInternalFuture<?> fut : futs)
-            fut.get();
-
-        Map<Integer, Integer> map = new HashMap<>();
-
-        for (Integer key : putKeys)
-            map.put(key, key);
+        fut.get(30_000);
 
-        checkData(map, grid(SRV_CNT).cache(null), SRV_CNT + CLIENT_CNT);
+        checkData(null, putKeys, grid(SRV_CNT).cache(null), SRV_CNT + CLIENT_CNT);
     }
 
     /**


Mime
View raw message