You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by iv...@apache.org on 2015/07/16 17:51:01 UTC
[21/36] incubator-ignite git commit: # ignite-901 client reconnect
support
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectFailoverTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectFailoverTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectFailoverTest.java
new file mode 100644
index 0000000..7cfc329
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectFailoverTest.java
@@ -0,0 +1,212 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.transactions.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+import static org.apache.ignite.transactions.TransactionIsolation.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectFailoverTest extends IgniteClientReconnectFailoverAbstractTest {
+ /** */
+ protected static final String ATOMIC_CACHE = "ATOMIC_CACHE";
+
+ /** */
+ protected static final String TX_CACHE = "TX_CACHE";
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration ccfg1 = new CacheConfiguration();
+
+ ccfg1.setName(ATOMIC_CACHE);
+ ccfg1.setBackups(1);
+ ccfg1.setAtomicityMode(ATOMIC);
+
+ CacheConfiguration ccfg2 = new CacheConfiguration();
+
+ ccfg2.setName(TX_CACHE);
+ ccfg2.setBackups(1);
+ ccfg2.setAtomicityMode(TRANSACTIONAL);
+
+ cfg.setCacheConfiguration(ccfg1, ccfg2);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectAtomicCache() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ final IgniteCache<Integer, Integer> cache = client.cache(ATOMIC_CACHE);
+
+ assertNotNull(cache);
+
+ assertEquals(ATOMIC, cache.getConfiguration(CacheConfiguration.class).getAtomicityMode());
+
+ reconnectFailover(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ TreeMap<Integer, Integer> map = new TreeMap<>();
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < 10; i++) {
+ Integer key = rnd.nextInt(0, 100_000);
+
+ cache.put(key, key);
+
+ assertEquals(key, cache.get(key));
+
+ map.put(key, key);
+ }
+
+ cache.putAll(map);
+
+ Map<Integer, Integer> res = cache.getAll(map.keySet());
+
+ assertEquals(map, res);
+
+ return null;
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectTxCache() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ final IgniteCache<Integer, Integer> cache = client.cache(TX_CACHE);
+
+ assertNotNull(cache);
+
+ assertEquals(TRANSACTIONAL, cache.getConfiguration(CacheConfiguration.class).getAtomicityMode());
+
+ final IgniteTransactions txs = client.transactions();
+
+ reconnectFailover(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ TreeMap<Integer, Integer> map = new TreeMap<>();
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < 5; i++) {
+ Integer key = rnd.nextInt(0, 100_000);
+
+ cache.put(key, key);
+
+ assertEquals(key, cache.get(key));
+
+ map.put(key, key);
+ }
+
+ for (TransactionConcurrency txConcurrency : TransactionConcurrency.values()) {
+ try (Transaction tx = txs.txStart(txConcurrency, REPEATABLE_READ)) {
+ for (Map.Entry<Integer, Integer> e : map.entrySet()) {
+ cache.put(e.getKey(), e.getValue());
+
+ assertNotNull(cache.get(e.getKey()));
+ }
+
+ tx.commit();
+ }
+ }
+
+ cache.putAll(map);
+
+ Map<Integer, Integer> res = cache.getAll(map.keySet());
+
+ assertEquals(map, res);
+
+ return null;
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectComputeApi() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ final IgniteCompute comp = client.compute();
+
+ reconnectFailover(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ comp.call(new DummyClosure());
+
+ comp.broadcast(new DummyClosure());
+
+ return null;
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectStreamerApi() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ reconnectFailover(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ stream(ATOMIC_CACHE);
+
+ stream(TX_CACHE);
+
+ return null;
+ }
+
+ private void stream(String cacheName) {
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ try (IgniteDataStreamer<Integer, Integer> streamer = client.dataStreamer(cacheName)) {
+ streamer.allowOverwrite(true);
+
+ streamer.perNodeBufferSize(10);
+
+ for (int i = 0; i < 100; i++)
+ streamer.addData(rnd.nextInt(100_000), 0);
+ }
+ }
+ });
+ }
+
+ /**
+ *
+ */
+ public static class DummyClosure implements IgniteCallable<Object> {
+ /** {@inheritDoc} */
+ @Override public Object call() throws Exception {
+ return 1;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java
new file mode 100644
index 0000000..31b4192
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectServicesTest.java
@@ -0,0 +1,260 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.processors.cache.distributed.near.*;
+import org.apache.ignite.internal.processors.service.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.services.*;
+import org.apache.ignite.testframework.*;
+
+import java.util.concurrent.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectServicesTest extends IgniteClientReconnectAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int serverCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int clientCount() {
+ return 1;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnect() throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ IgniteServices services = client.services();
+
+ services.deployClusterSingleton("testReconnect", new TestServiceImpl());
+
+ TestService srvc = services.serviceProxy("testReconnect", TestService.class, false);
+
+ assertNotNull(srvc);
+
+ long topVer = grid(0).cluster().topologyVersion();
+
+ assertEquals((Object)topVer, srvc.test());
+
+ Ignite srv = clientRouter(client);
+
+ reconnectClientNode(client, srv, null);
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ DummyService.exeLatch("testReconnect2", latch);
+
+ services.deployClusterSingleton("testReconnect2", new DummyService());
+
+ assertTrue(latch.await(5000, TimeUnit.MILLISECONDS));
+
+ assertEquals((Object)(topVer + 2), srvc.test());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testServiceRemove() throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ Ignite srv = clientRouter(client);
+
+ IgniteServices clnServices = client.services();
+
+ final IgniteServices srvServices = srv.services();
+
+ srvServices.deployClusterSingleton("testServiceRemove", new TestServiceImpl());
+
+ final TestService srvc = clnServices.serviceProxy("testServiceRemove", TestService.class, false);
+
+ assertNotNull(srvc);
+
+ assertNotNull(srvc.test());
+
+ reconnectClientNode(client, srv, new Runnable() {
+ @Override public void run() {
+ srvServices.cancel("testServiceRemove");
+ }
+ });
+
+ GridTestUtils.assertThrows(log, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ return srvc.test();
+ }
+ }, IgniteException.class, null);
+
+ clnServices.deployClusterSingleton("testServiceRemove", new TestServiceImpl());
+
+ TestService newSrvc = clnServices.serviceProxy("testServiceRemove", TestService.class, false);
+
+ assertNotNull(newSrvc);
+ assertNotNull(newSrvc.test());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectInDeploying() throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ final IgniteServices services = client.services();
+
+ Ignite srv = clientRouter(client);
+
+ BlockTpcCommunicationSpi commSpi = commSpi(srv);
+
+ commSpi.blockMessage(GridNearTxPrepareResponse.class);
+
+ final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ try {
+ services.deployClusterSingleton("testReconnectInDeploying", new TestServiceImpl());
+ }
+ catch (IgniteClientDisconnectedException e) {
+ checkAndWait(e);
+
+ return true;
+ }
+
+ return false;
+ }
+ });
+
+ // Check that client waiting operation.
+ GridTestUtils.assertThrows(log, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ return fut.get(200);
+ }
+ }, IgniteFutureTimeoutCheckedException.class, null);
+
+ assertNotDone(fut);
+
+ commSpi.unblockMessage();
+
+ reconnectClientNode(client, srv, null);
+
+ assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectInProgress() throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ final IgniteServices services = client.services();
+
+ final Ignite srv = clientRouter(client);
+
+ services.deployClusterSingleton("testReconnectInProgress", new TestServiceImpl());
+
+ final TestService srvc = services.serviceProxy("testReconnectInProgress", TestService.class, false);
+
+ assertNotNull(srvc);
+
+ BlockTpcCommunicationSpi commSpi = commSpi(srv);
+
+ commSpi.blockMessage(GridJobExecuteResponse.class);
+
+ final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ try {
+ srvc.test();
+ }
+ catch (IgniteClientDisconnectedException e) {
+ checkAndWait(e);
+
+ return true;
+ }
+
+ return false;
+ }
+ });
+
+ // Check that client waiting operation.
+ GridTestUtils.assertThrows(log, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ return fut.get(200);
+ }
+ }, IgniteFutureTimeoutCheckedException.class, null);
+
+ assertNotDone(fut);
+
+ commSpi.unblockMessage();
+
+ reconnectClientNode(client, srv, null);
+
+ assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+ }
+
+ /**
+ *
+ */
+ public static interface TestService {
+ /**
+ * @return Topology version.
+ */
+ public Long test();
+ }
+
+ /**
+ *
+ */
+ public static class TestServiceImpl implements Service, TestService {
+ /** */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** {@inheritDoc} */
+ @Override public void cancel(ServiceContext ctx) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void init(ServiceContext ctx) throws Exception {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void execute(ServiceContext ctx) throws Exception {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public Long test() {
+ assertFalse(ignite.cluster().localNode().isClient());
+
+ return ignite.cluster().topologyVersion();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStopTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStopTest.java
new file mode 100644
index 0000000..98c3d0f
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStopTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.lang.*;
+
+import java.util.concurrent.*;
+
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectStopTest extends IgniteClientReconnectAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected int serverCount() {
+ return 1;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStopWhenDisconnected() throws Exception {
+ clientMode = true;
+
+ Ignite client = startGrid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ Ignite srv = clientRouter(client);
+
+ TestTcpDiscoverySpi srvSpi = spi(srv);
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+ final CountDownLatch reconnectLatch = new CountDownLatch(1);
+
+ final TestTcpDiscoverySpi clientSpi = spi(client);
+
+ log.info("Block reconnect.");
+
+ clientSpi.writeLatch = new CountDownLatch(1);
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ info("Disconnected: " + evt);
+
+ disconnectLatch.countDown();
+ } else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+ info("Reconnected: " + evt);
+
+ reconnectLatch.countDown();
+ }
+
+ return true;
+ }
+ }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED);
+
+ srvSpi.failNode(client.cluster().localNode().id(), null);
+
+ waitReconnectEvent(disconnectLatch);
+
+ IgniteFuture<?> reconnectFut = null;
+
+ try {
+ client.getOrCreateCache(new CacheConfiguration<>());
+
+ fail();
+ }
+ catch (IgniteClientDisconnectedException e) {
+ log.info("Expected operation exception: " + e);
+
+ reconnectFut = e.reconnectFuture();
+ }
+
+ assertNotNull(reconnectFut);
+
+ client.close();
+
+ try {
+ reconnectFut.get();
+
+ fail();
+ }
+ catch (IgniteException e) {
+ log.info("Expected reconnect exception: " + e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStreamerTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStreamerTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStreamerTest.java
new file mode 100644
index 0000000..a4cf77f
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectStreamerTest.java
@@ -0,0 +1,233 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.processors.datastreamer.*;
+import org.apache.ignite.internal.util.lang.*;
+import org.apache.ignite.testframework.*;
+
+import javax.cache.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+import static org.apache.ignite.cache.CacheMode.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectStreamerTest extends IgniteClientReconnectAbstractTest {
+ /** */
+ public static final String CACHE_NAME = "streamer";
+
+ /** {@inheritDoc} */
+ @Override protected int serverCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int clientCount() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration<Integer, Integer> ccfg = new CacheConfiguration<Integer, Integer>(CACHE_NAME)
+ .setAtomicityMode(ATOMIC)
+ .setCacheMode(PARTITIONED);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStreamerReconnect() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ Ignite srv = clientRouter(client);
+
+ final IgniteCache<Object, Object> srvCache = srv.cache(CACHE_NAME);
+
+ IgniteDataStreamer<Integer, Integer> streamer = client.dataStreamer(CACHE_NAME);
+
+ for (int i = 0; i < 50; i++)
+ streamer.addData(i, i);
+
+ streamer.flush();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return srvCache.localSize() == 50;
+ }
+ }, 2000L);
+
+ assertEquals(50, srvCache.localSize());
+
+ reconnectClientNode(client, srv, new Runnable() {
+ @Override public void run() {
+ try {
+ client.dataStreamer(CACHE_NAME);
+
+ fail();
+ }
+ catch (IgniteClientDisconnectedException e) {
+ assertNotNull(e.reconnectFuture());
+ }
+ }
+ });
+
+ checkStreamerClosed(streamer);
+
+ streamer = client.dataStreamer(CACHE_NAME);
+
+ for (int i = 50; i < 100; i++)
+ streamer.addData(i, i);
+
+ streamer.flush();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return srvCache.localSize() == 100;
+ }
+ }, 2000L);
+
+ assertEquals(100, srvCache.localSize());
+
+ streamer.close();
+
+ streamer.future().get(2, TimeUnit.SECONDS);
+
+ srvCache.removeAll();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testStreamerReconnectInProgress() throws Exception {
+ Ignite client = grid(serverCount());
+
+ assertTrue(client.cluster().localNode().isClient());
+
+ Ignite srv = clientRouter(client);
+
+ final IgniteCache<Object, Object> srvCache = srv.cache(CACHE_NAME);
+
+ final IgniteDataStreamer<Integer, Integer> streamer = client.dataStreamer(CACHE_NAME);
+
+ BlockTpcCommunicationSpi commSpi = commSpi(srv);
+
+ commSpi.blockMessage(DataStreamerResponse.class);
+
+ final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ try {
+ for (int i = 0; i < 50; i++)
+ streamer.addData(i, i);
+
+ streamer.flush();
+ }
+ catch (CacheException e) {
+ checkAndWait(e);
+
+ return true;
+ }
+ finally {
+ streamer.close();
+ }
+
+ return false;
+ }
+ });
+
+ // Check that client waiting operation.
+ GridTestUtils.assertThrows(log, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ return fut.get(200);
+ }
+ }, IgniteFutureTimeoutCheckedException.class, null);
+
+ assertNotDone(fut);
+
+ commSpi.unblockMessage();
+
+ reconnectClientNode(client, srv, null);
+
+ assertTrue((Boolean)fut.get(2, TimeUnit.SECONDS));
+
+ checkStreamerClosed(streamer);
+
+ IgniteDataStreamer<Integer, Integer> streamer2 = client.dataStreamer(CACHE_NAME);
+
+ for (int i = 0; i < 50; i++)
+ streamer2.addData(i, i);
+
+ streamer2.close();
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return srvCache.localSize() == 50;
+ }
+ }, 2000L);
+
+ assertEquals(50, srvCache.localSize());
+ }
+
+ /**
+ * @param streamer Streamer.
+ */
+ private void checkStreamerClosed(IgniteDataStreamer<Integer, Integer> streamer) {
+ try {
+ streamer.addData(100, 100);
+
+ fail();
+ }
+ catch (CacheException e) {
+ checkAndWait(e);
+ }
+
+ try {
+ streamer.flush();
+
+ fail();
+ }
+ catch (CacheException e) {
+ checkAndWait(e);
+ }
+
+ try {
+ streamer.future().get();
+
+ fail();
+ }
+ catch (CacheException e) {
+ checkAndWait(e);
+ }
+
+ streamer.tryFlush();
+
+ streamer.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/IgniteSlowClientDetectionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteSlowClientDetectionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteSlowClientDetectionSelfTest.java
index 27c2a61..a392245 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteSlowClientDetectionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteSlowClientDetectionSelfTest.java
@@ -62,6 +62,7 @@ public class IgniteSlowClientDetectionSelfTest extends GridCommonAbstractTest {
IgniteConfiguration cfg = super.getConfiguration(gridName);
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setClientReconnectDisabled(true);
if (getTestGridName(nodeCount() - 1).equals(gridName) || getTestGridName(nodeCount() - 2).equals(gridName))
cfg.setClientMode(true);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManagerStopSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManagerStopSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManagerStopSelfTest.java
index 9780080..62f5d41 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManagerStopSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/deployment/GridDeploymentManagerStopSelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.managers.deployment;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.resource.*;
+import org.apache.ignite.lang.*;
import org.apache.ignite.marshaller.jdk.*;
import org.apache.ignite.spi.*;
import org.apache.ignite.spi.deployment.*;
@@ -95,5 +96,11 @@ public class GridDeploymentManagerStopSelfTest extends GridCommonAbstractTest {
/** {@inheritDoc} */
@Override public boolean unregister(String rsrcName) { return false; }
+
+ /** {@inheritDoc} */
+ @Override public void onClientDisconnected(IgniteFuture<?> reconnectFut) { /* No-op. */ }
+
+ /** {@inheritDoc} */
+ @Override public void onClientReconnected(boolean clusterRestarted) { /* No-op. */ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
index 074f6ff..9c30f23 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractStopBusySelfTest.java
@@ -263,7 +263,7 @@ public abstract class IgniteCacheAbstractStopBusySelfTest extends GridCommonAbst
e.printStackTrace(pw);
- assertTrue(sw.toString().contains("grid is stopping"));
+ assertTrue(sw.toString().contains("node is stopping"));
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheDynamicStopSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheDynamicStopSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheDynamicStopSelfTest.java
index 071341e..8703d32 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheDynamicStopSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheDynamicStopSelfTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.cache;
import org.apache.ignite.IgniteDataStreamer;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.IgniteInternalFuture;
@@ -27,7 +26,7 @@ import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.testframework.*;
import org.apache.ignite.testframework.junits.common.*;
-import javax.cache.Cache;
+import javax.cache.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -89,7 +88,8 @@ public class IgniteCacheDynamicStopSelfTest extends GridCommonAbstractTest {
@Override public void apply(IgniteFuture<?> f) {
try {
f.get();
- } catch (IgniteException ignore) {
+ }
+ catch (CacheException ignore) {
// This may be debugged.
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
index af3ea9d..30bf5dc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteTxExceptionAbstractSelfTest.java
@@ -22,6 +22,7 @@ import org.apache.ignite.cluster.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.distributed.near.*;
+import org.apache.ignite.internal.transactions.*;
import org.apache.ignite.spi.*;
import org.apache.ignite.spi.indexing.*;
import org.apache.ignite.testframework.*;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSystemTransactionsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSystemTransactionsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSystemTransactionsSelfTest.java
index d78add6..53404cc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSystemTransactionsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheSystemTransactionsSelfTest.java
@@ -75,7 +75,7 @@ public class IgniteCacheSystemTransactionsSelfTest extends GridCacheAbstractSelf
jcache.get("1");
jcache.put("1", "11");
- GridCacheAdapter<Object, Object> utilityCache = ignite.context().cache().utilityCache();
+ IgniteInternalCache<Object, Object> utilityCache = ignite.context().cache().utilityCache();
utilityCache.getAndPutIfAbsent("2", "2");
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedInvalidateSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedInvalidateSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedInvalidateSelfTest.java
index 19e40bf..7a2e8b3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedInvalidateSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/replicated/GridCacheReplicatedInvalidateSelfTest.java
@@ -220,7 +220,8 @@ public class GridCacheReplicatedInvalidateSelfTest extends GridCommonAbstractTes
Object msg0 = ((GridIoMessage)msg).message();
if (!(msg0 instanceof GridClockDeltaSnapshotMessage)) {
- info("Sending message [locNodeId=" + getLocalNodeId() + ", destNodeId= " + destNode.id()
+ info("Sending message [locNodeId=" + ignite.cluster().localNode().id() +
+ ", destNodeId= " + destNode.id()
+ ", msg=" + msg + ']');
synchronized (msgCntMap) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
index e9d7a45..9a883b3 100644
--- a/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
+++ b/modules/core/src/test/java/org/apache/ignite/loadtests/hashmap/GridCacheTestContext.java
@@ -55,8 +55,8 @@ public class GridCacheTestContext<K, V> extends GridCacheContext<K, V> {
new GridCacheDeploymentManager<K, V>(),
new GridCachePartitionExchangeManager<K, V>(),
new GridCacheIoManager(),
- null,
- new CacheNoopJtaManager()
+ new CacheNoopJtaManager(),
+ null
),
defaultCacheConfiguration(),
CacheType.USER,
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index ec6a526..63db0c1 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@ -111,6 +111,9 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
/** */
private IgniteInClosure2X<TcpDiscoveryAbstractMessage, Socket> afterWrite;
+ /** */
+ private boolean reconnectDisabled;
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
@@ -159,6 +162,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
disco.setJoinTimeout(joinTimeout);
disco.setNetworkTimeout(netTimeout);
+ disco.setClientReconnectDisabled(reconnectDisabled);
+
disco.afterWrite(afterWrite);
cfg.setDiscoverySpi(disco);
@@ -524,7 +529,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
TestTcpDiscoverySpi spi = ((TestTcpDiscoverySpi)ignite.configuration().getDiscoverySpi());
- spi.pauseAll();
+ spi.pauseAll(false);
try {
spi.brakeConnection();
@@ -568,7 +573,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
TestTcpDiscoverySpi spi = ((TestTcpDiscoverySpi)ignite.configuration().getDiscoverySpi());
- spi.pauseAll();
+ spi.pauseAll(false);
try {
spi.brakeConnection();
@@ -606,7 +611,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
attachListeners(2, 2);
- ((TestTcpDiscoverySpi)G.ignite("client-1").configuration().getDiscoverySpi()).pauseAll();
+ ((TestTcpDiscoverySpi)G.ignite("client-1").configuration().getDiscoverySpi()).pauseAll(true);
stopGrid("server-2");
@@ -633,6 +638,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
public void testClientSegmentation() throws Exception {
clientsPerSrv = 1;
+ reconnectDisabled = true;
+
startServerNodes(3);
startClientNodes(3);
@@ -656,6 +663,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
final TcpDiscoverySpi disco = (TcpDiscoverySpi)G.ignite("client-2").configuration().getDiscoverySpi();
try {
+ log.info("Fail server: " + 2);
+
failServer(2);
await(srvFailedLatch);
@@ -886,8 +895,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
try {
startClientNodes(1);
- assertEquals(G.ignite("server-0").cluster().localNode().id(), ((TcpDiscoveryNode)G.ignite("client-0")
- .cluster().localNode()).clientRouterNodeId());
+ assertEquals(G.ignite("server-0").cluster().localNode().id(),
+ ((TcpDiscoveryNode) G.ignite("client-0").cluster().localNode()).clientRouterNodeId());
checkNodes(2, 1);
@@ -1206,6 +1215,528 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ * @throws Exception If failed.
+ */
+ public void testReconnectAfterFail() throws Exception {
+ reconnectAfterFail(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectAfterFailTopologyChanged() throws Exception {
+ reconnectAfterFail(true);
+ }
+
+ /**
+ * @param changeTop If {@code true} topology is changed after client disconnects.
+ * @throws Exception If failed.
+ */
+ private void reconnectAfterFail(final boolean changeTop) throws Exception {
+ startServerNodes(1);
+
+ startClientNodes(1);
+
+ Ignite srv = G.ignite("server-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)srv.configuration().getDiscoverySpi());
+
+ Ignite client = G.ignite("client-0");
+
+ final ClusterNode clientNode = client.cluster().localNode();
+
+ final UUID clientId = clientNode.id();
+
+ final TestTcpDiscoverySpi clientSpi = ((TestTcpDiscoverySpi)client.configuration().getDiscoverySpi());
+
+ assertEquals(2L, clientNode.order());
+
+ final CountDownLatch failLatch = new CountDownLatch(1);
+
+ final CountDownLatch joinLatch = new CountDownLatch(1);
+
+ srv.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ info("Server event: " + evt);
+
+ DiscoveryEvent evt0 = (DiscoveryEvent)evt;
+
+ if (evt0.eventNode().id().equals(clientId) && (evt.type() == EVT_NODE_FAILED)) {
+ if (evt.type() == EVT_NODE_FAILED)
+ failLatch.countDown();
+ }
+ else if (evt.type() == EVT_NODE_JOINED) {
+ TcpDiscoveryNode node = (TcpDiscoveryNode)evt0.eventNode();
+
+ if ("client-0".equals(node.attribute(IgniteNodeAttributes.ATTR_GRID_NAME))) {
+ assertEquals(changeTop ? 5L : 4L, node.order());
+
+ joinLatch.countDown();
+ }
+ }
+
+ return true;
+ }
+ }, EVT_NODE_FAILED, EVT_NODE_JOINED);
+
+ final CountDownLatch reconnectLatch = new CountDownLatch(1);
+
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ info("Client event: " + evt);
+
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ assertEquals(1, reconnectLatch.getCount());
+
+ disconnectLatch.countDown();
+
+ if (changeTop)
+ clientSpi.pauseAll(false);
+ }
+ else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+ assertEquals(0, disconnectLatch.getCount());
+
+ reconnectLatch.countDown();
+ }
+
+ return true;
+ }
+ }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED);
+
+ srvSpi.failNode(client.cluster().localNode().id(), null);
+
+ if (changeTop) {
+ Ignite g = startGrid("server-" + srvIdx.getAndIncrement());
+
+ srvNodeIds.add(g.cluster().localNode().id());
+
+ clientSpi.resumeAll();
+ }
+
+ assertTrue(disconnectLatch.await(5000, MILLISECONDS));
+ assertTrue(reconnectLatch.await(5000, MILLISECONDS));
+ assertTrue(failLatch.await(5000, MILLISECONDS));
+ assertTrue(joinLatch.await(5000, MILLISECONDS));
+
+ long topVer = changeTop ? 5L : 4L;
+
+ assertEquals(topVer, client.cluster().localNode().order());
+
+ assertEquals(topVer, client.cluster().topologyVersion());
+
+ Collection<ClusterNode> clientTop = client.cluster().topology(topVer);
+
+ assertEquals(changeTop ? 3 : 2, clientTop.size());
+
+ clientNodeIds.remove(clientId);
+
+ clientNodeIds.add(client.cluster().localNode().id());
+
+ checkNodes(changeTop ? 2 : 1, 1);
+
+ Ignite g = startGrid("server-" + srvIdx.getAndIncrement());
+
+ srvNodeIds.add(g.cluster().localNode().id());
+
+ checkNodes(changeTop ? 3 : 2, 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectAfterFailConcurrentJoin() throws Exception {
+ startServerNodes(1);
+
+ startClientNodes(1);
+
+ Ignite srv = G.ignite("server-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)srv.configuration().getDiscoverySpi());
+
+ Ignite client = G.ignite("client-0");
+
+ final ClusterNode clientNode = client.cluster().localNode();
+
+ assertEquals(2L, clientNode.order());
+
+ final CountDownLatch reconnectLatch = new CountDownLatch(1);
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ info("Client event: " + evt);
+
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ assertEquals(1, reconnectLatch.getCount());
+
+ disconnectLatch.countDown();
+ }
+ else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+ assertEquals(0, disconnectLatch.getCount());
+
+ reconnectLatch.countDown();
+ }
+
+ return true;
+ }
+ }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED);
+
+ final int CLIENTS = 20;
+
+ clientsPerSrv = CLIENTS + 1;
+
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ latch.await();
+
+ Ignite g = startGrid("client-" + clientIdx.getAndIncrement());
+
+ clientNodeIds.add(g.cluster().localNode().id());
+
+ return null;
+ }
+ }, CLIENTS, "start-client");
+
+ srvSpi.failNode(client.cluster().localNode().id(), null);
+
+ latch.countDown();
+
+ assertTrue(disconnectLatch.await(10_000, MILLISECONDS));
+ assertTrue(reconnectLatch.await(10_000, MILLISECONDS));
+
+ clientNodeIds.add(client.cluster().localNode().id());
+
+ fut.get();
+
+ checkNodes(1, CLIENTS + 1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testClientFailReconnectDisabled() throws Exception {
+ reconnectDisabled = true;
+
+ startServerNodes(1);
+
+ startClientNodes(1);
+
+ Ignite srv = G.ignite("server-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)srv.configuration().getDiscoverySpi());
+
+ Ignite client = G.ignite("client-0");
+
+ final CountDownLatch segmentedLatch = new CountDownLatch(1);
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ if (evt.type() == EVT_NODE_SEGMENTED)
+ segmentedLatch.countDown();
+
+ return false;
+ }
+ }, EVT_NODE_SEGMENTED);
+
+ srvFailedLatch = new CountDownLatch(1);
+
+ attachListeners(1, 0);
+
+ log.info("Fail client node.");
+
+ srvSpi.failNode(client.cluster().localNode().id(), null);
+
+ assertTrue(srvFailedLatch.await(5000, MILLISECONDS));
+ assertTrue(segmentedLatch.await(5000, MILLISECONDS));
+
+ checkNodes(1, 0);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectSegmentedAfterJoinTimeoutServerFailed() throws Exception {
+ reconnectSegmentedAfterJoinTimeout(true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectSegmentedAfterJoinTimeoutNetworkError() throws Exception {
+ reconnectSegmentedAfterJoinTimeout(false);
+ }
+
+ /**
+ * @param failSrv If {@code true} fails server, otherwise server does not send join message.
+ * @throws Exception If failed.
+ */
+ private void reconnectSegmentedAfterJoinTimeout(boolean failSrv) throws Exception {
+ netTimeout = 4000;
+ joinTimeout = 5000;
+
+ startServerNodes(1);
+
+ startClientNodes(1);
+
+ final Ignite srv = G.ignite("server-0");
+ Ignite client = G.ignite("client-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)srv.configuration().getDiscoverySpi());
+ TestTcpDiscoverySpi clientSpi = ((TestTcpDiscoverySpi)client.configuration().getDiscoverySpi());
+
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+ final CountDownLatch segmentedLatch = new CountDownLatch(1);
+ final AtomicBoolean err = new AtomicBoolean(false);
+
+ if (!failSrv) {
+ srvFailedLatch = new CountDownLatch(1);
+
+ attachListeners(1, 0);
+ }
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ log.info("Disconnected event.");
+
+ assertEquals(1, segmentedLatch.getCount());
+ assertEquals(1, disconnectLatch.getCount());
+ assertFalse(err.get());
+
+ disconnectLatch.countDown();
+ }
+ else if (evt.type() == EVT_NODE_SEGMENTED) {
+ log.info("Segmented event.");
+
+ assertEquals(1, segmentedLatch.getCount());
+ assertEquals(0, disconnectLatch.getCount());
+ assertFalse(err.get());
+
+ segmentedLatch.countDown();
+ }
+ else {
+ log.error("Unexpected event: " + evt);
+
+ err.set(true);
+ }
+
+ return true;
+ }
+ }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED, EVT_NODE_SEGMENTED);
+
+ if (failSrv) {
+ log.info("Fail server.");
+
+ failServer(0);
+ }
+ else {
+ log.info("Fail client connection.");
+
+ srvSpi.failClientReconnect.set(1_000_000);
+ srvSpi.skipNodeAdded = true;
+
+ clientSpi.brakeConnection();
+ }
+
+ assertTrue(disconnectLatch.await(10_000, MILLISECONDS));
+
+ assertTrue(segmentedLatch.await(10_000, MILLISECONDS));
+
+ waitSegmented(client);
+
+ assertFalse(err.get());
+
+ if (!failSrv) {
+ await(srvFailedLatch);
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return srv.cluster().nodes().size() == 1;
+ }
+ }, 10_000);
+
+ checkNodes(1, 0);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectClusterRestart() throws Exception {
+ netTimeout = 3000;
+ joinTimeout = 60_000;
+
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+ final CountDownLatch reconnectLatch = new CountDownLatch(1);
+ final AtomicBoolean err = new AtomicBoolean(false);
+
+ startServerNodes(1);
+
+ startClientNodes(1);
+
+ Ignite srv = G.ignite("server-0");
+ Ignite client = G.ignite("client-0");
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ log.info("Disconnected event.");
+
+ assertEquals(1, reconnectLatch.getCount());
+ assertEquals(1, disconnectLatch.getCount());
+
+ disconnectLatch.countDown();
+ }
+ else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+ log.info("Reconnected event.");
+
+ assertEquals(1, reconnectLatch.getCount());
+ assertEquals(0, disconnectLatch.getCount());
+ assertFalse(err.get());
+
+ reconnectLatch.countDown();
+ }
+ else {
+ log.error("Unexpected event: " + evt);
+
+ err.set(true);
+ }
+
+ return true;
+ }
+ }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED, EVT_NODE_SEGMENTED);
+
+ log.info("Stop server.");
+
+ srv.close();
+
+ assertTrue(disconnectLatch.await(10_000, MILLISECONDS));
+
+ srvNodeIds.clear();
+ srvIdx.set(0);
+
+ Thread.sleep(3000);
+
+ log.info("Restart server.");
+
+ startServerNodes(1);
+
+ assertTrue(reconnectLatch.await(10_000, MILLISECONDS));
+
+ clientNodeIds.clear();
+ clientNodeIds.add(client.cluster().localNode().id());
+
+ checkNodes(1, 1);
+
+ assertFalse(err.get());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDisconnectAfterNetworkTimeout() throws Exception {
+ netTimeout = 5000;
+ joinTimeout = 60_000;
+ maxMissedClientHbs = 2;
+
+ startServerNodes(1);
+
+ startClientNodes(1);
+
+ final Ignite srv = G.ignite("server-0");
+ Ignite client = G.ignite("client-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)srv.configuration().getDiscoverySpi());
+ TestTcpDiscoverySpi clientSpi = ((TestTcpDiscoverySpi)client.configuration().getDiscoverySpi());
+
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+ final CountDownLatch reconnectLatch = new CountDownLatch(1);
+ final AtomicBoolean err = new AtomicBoolean(false);
+
+ client.events().localListen(new IgnitePredicate<Event>() {
+ @Override
+ public boolean apply(Event evt) {
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ log.info("Disconnected event.");
+
+ assertEquals(1, reconnectLatch.getCount());
+ assertEquals(1, disconnectLatch.getCount());
+ assertFalse(err.get());
+
+ disconnectLatch.countDown();
+ }
+ else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+ log.info("Reconnected event.");
+
+ assertEquals(1, reconnectLatch.getCount());
+ assertEquals(0, disconnectLatch.getCount());
+ assertFalse(err.get());
+
+ reconnectLatch.countDown();
+ }
+ else {
+ log.error("Unexpected event: " + evt);
+
+ err.set(true);
+ }
+
+ return true;
+ }
+ }, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED, EVT_NODE_SEGMENTED);
+
+ log.info("Fail client connection1.");
+
+ srvSpi.failClientReconnect.set(1_000_000);
+ srvSpi.skipNodeAdded = true;
+
+ clientSpi.brakeConnection();
+
+ assertTrue(disconnectLatch.await(10_000, MILLISECONDS));
+
+ log.info("Fail client connection2.");
+
+ srvSpi.failClientReconnect.set(0);
+ srvSpi.skipNodeAdded = false;
+
+ clientSpi.brakeConnection();
+
+ assertTrue(reconnectLatch.await(10_000, MILLISECONDS));
+
+ clientNodeIds.clear();
+
+ clientNodeIds.add(client.cluster().localNode().id());
+
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override
+ public boolean apply() {
+ return srv.cluster().nodes().size() == 2;
+ }
+ }, 10_000);
+
+ checkNodes(1, 1);
+
+ assertFalse(err.get());
+ }
+
+ /**
+ * @param ignite Ignite.
+ * @throws Exception If failed.
+ */
+ private void waitSegmented(final Ignite ignite) throws Exception {
+ GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ return IgniteState.STOPPED_ON_SEGMENTATION == Ignition.state(ignite.name());
+ }
+ }, 5000);
+
+ assertEquals(IgniteState.STOPPED_ON_SEGMENTATION, Ignition.state(ignite.name()));
+ }
+
+ /**
* @param clientIdx Client index.
* @param srvIdx Server index.
* @throws Exception In case of error.
@@ -1401,7 +1932,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
private void checkRemoteNodes(Ignite ignite, int expCnt) {
Collection<ClusterNode> nodes = ignite.cluster().forRemotes().nodes();
- assertEquals(expCnt, nodes.size());
+ assertEquals("Unexpected state for node: " + ignite.name(), expCnt, nodes.size());
for (ClusterNode node : nodes) {
UUID id = node.id();
@@ -1420,7 +1951,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
* @throws InterruptedException If interrupted.
*/
private void await(CountDownLatch latch) throws InterruptedException {
- assertTrue("Latch count: " + latch.getCount(), latch.await(10000, MILLISECONDS));
+ assertTrue("Latch count: " + latch.getCount(), latch.await(10_000, MILLISECONDS));
}
/**
@@ -1471,6 +2002,9 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
/** */
private volatile String delayJoinAckFor;
+ /** */
+ private volatile boolean skipNodeAdded;
+
/**
* @param lock Lock.
*/
@@ -1543,6 +2077,13 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
boolean fail = false;
+ if (skipNodeAdded &&
+ (msg instanceof TcpDiscoveryNodeAddedMessage || msg instanceof TcpDiscoveryNodeAddFinishedMessage)) {
+ log.info("Skip message: " + msg);
+
+ return;
+ }
+
if (msg instanceof TcpDiscoveryNodeAddedMessage)
fail = failNodeAdded.getAndDecrement() > 0;
else if (msg instanceof TcpDiscoveryNodeAddFinishedMessage)
@@ -1577,12 +2118,13 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
- *
+ * @param suspend If {@code true} suspends worker threads.
*/
- public void pauseAll() {
+ public void pauseAll(boolean suspend) {
pauseResumeOperation(true, openSockLock, writeLock);
- impl.workerThread().suspend();
+ if (suspend)
+ impl.workerThread().suspend();
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
index 159c451..dacbf55 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/multijvm/IgniteClusterProcessProxy.java
@@ -317,4 +317,9 @@ public class IgniteClusterProcessProxy implements IgniteClusterEx {
@Override public ClusterMetrics metrics() throws IgniteException {
throw new UnsupportedOperationException("Operation is not supported yet.");
}
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IgniteFuture<?> clientReconnectFuture() {
+ throw new UnsupportedOperationException("Operation is not supported yet.");
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteClientReconnectTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteClientReconnectTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteClientReconnectTestSuite.java
new file mode 100644
index 0000000..66c9835
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteClientReconnectTestSuite.java
@@ -0,0 +1,48 @@
+/*
+ * 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.testsuites;
+
+import junit.framework.*;
+import org.apache.ignite.internal.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectTestSuite extends TestSuite {
+ /**
+ * @return Test suite.
+ * @throws Exception In case of error.
+ */
+ public static TestSuite suite() throws Exception {
+ TestSuite suite = new TestSuite("Ignite Client Reconnect Test Suite");
+
+ suite.addTestSuite(IgniteClientReconnectStopTest.class);
+ suite.addTestSuite(IgniteClientReconnectApiExceptionTest.class);
+ suite.addTestSuite(IgniteClientReconnectDiscoveryStateTest.class);
+ suite.addTestSuite(IgniteClientReconnectCacheTest.class);
+ suite.addTestSuite(IgniteClientReconnectContinuousProcessorTest.class);
+ suite.addTestSuite(IgniteClientReconnectComputeTest.class);
+ suite.addTestSuite(IgniteClientReconnectAtomicsTest.class);
+ suite.addTestSuite(IgniteClientReconnectCollectionsTest.class);
+ suite.addTestSuite(IgniteClientReconnectServicesTest.class);
+ suite.addTestSuite(IgniteClientReconnectStreamerTest.class);
+ suite.addTestSuite(IgniteClientReconnectFailoverTest.class);
+
+ return suite;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index 06c0961..c76dbe7 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -1439,6 +1439,11 @@ public class IgniteH2Indexing implements GridQueryIndexing {
fut.get();
}
+ /** {@inheritDoc} */
+ @Override public void onDisconnected(IgniteFuture<?> reconnectFut) {
+ rdcQryExec.onDisconnected(reconnectFut);
+ }
+
/**
* Wrapper to store connection and flag is schema set or not.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
index af29647..2b2996d 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMergeIndex.java
@@ -26,6 +26,7 @@ import org.h2.table.*;
import org.jetbrains.annotations.*;
import org.jsr166.*;
+import javax.cache.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
@@ -40,7 +41,7 @@ public abstract class GridMergeIndex extends BaseIndex {
private static final int MAX_FETCH_SIZE = getInteger(IGNITE_SQL_MERGE_TABLE_MAX_SIZE, 10_000);
/** All rows number. */
- private final AtomicInteger expectedRowsCnt = new AtomicInteger(0);
+ private final AtomicInteger expRowsCnt = new AtomicInteger(0);
/** Remaining rows per source node ID. */
private final ConcurrentMap<UUID, Counter> remainingRows = new ConcurrentHashMap8<>();
@@ -75,8 +76,8 @@ public abstract class GridMergeIndex extends BaseIndex {
}
/** {@inheritDoc} */
- @Override public long getRowCount(Session session) {
- return expectedRowsCnt.get();
+ @Override public long getRowCount(Session ses) {
+ return expRowsCnt.get();
}
/** {@inheritDoc} */
@@ -93,6 +94,23 @@ public abstract class GridMergeIndex extends BaseIndex {
}
/**
+ * @param e Error.
+ */
+ public void fail(final CacheException e) {
+ for (UUID nodeId0 : remainingRows.keySet()) {
+ addPage0(new GridResultPage(null, nodeId0, null) {
+ @Override public boolean isFail() {
+ return true;
+ }
+
+ @Override public void fetchNextPage() {
+ throw e;
+ }
+ });
+ }
+ }
+
+ /**
* @param nodeId Node ID.
*/
public void fail(UUID nodeId) {
@@ -120,7 +138,7 @@ public abstract class GridMergeIndex extends BaseIndex {
assert !cnt.initialized : "Counter is already initialized.";
cnt.addAndGet(allRows);
- expectedRowsCnt.addAndGet(allRows);
+ expRowsCnt.addAndGet(allRows);
// We need this separate flag to handle case when the first source contains only one page
// and it will signal that all remaining counters are zero and fetch is finished.
@@ -162,7 +180,7 @@ public abstract class GridMergeIndex extends BaseIndex {
}
/** {@inheritDoc} */
- @Override public Cursor find(Session session, SearchRow first, SearchRow last) {
+ @Override public Cursor find(Session ses, SearchRow first, SearchRow last) {
if (fetched == null)
throw new IgniteException("Fetched result set was too large.");
@@ -176,7 +194,7 @@ public abstract class GridMergeIndex extends BaseIndex {
* @return {@code true} If we have fetched all the remote rows.
*/
public boolean fetchedAll() {
- return fetchedCnt == expectedRowsCnt.get();
+ return fetchedCnt == expRowsCnt.get();
}
/**
@@ -200,32 +218,32 @@ public abstract class GridMergeIndex extends BaseIndex {
}
/** {@inheritDoc} */
- @Override public void close(Session session) {
+ @Override public void close(Session ses) {
// No-op.
}
/** {@inheritDoc} */
- @Override public void add(Session session, Row row) {
+ @Override public void add(Session ses, Row row) {
throw DbException.getUnsupportedException("add");
}
/** {@inheritDoc} */
- @Override public void remove(Session session, Row row) {
+ @Override public void remove(Session ses, Row row) {
throw DbException.getUnsupportedException("remove row");
}
/** {@inheritDoc} */
- @Override public double getCost(Session session, int[] masks, TableFilter filter, SortOrder sortOrder) {
+ @Override public double getCost(Session ses, int[] masks, TableFilter filter, SortOrder sortOrder) {
return getRowCountApproximation() + Constants.COST_ROW_OFFSET;
}
/** {@inheritDoc} */
- @Override public void remove(Session session) {
+ @Override public void remove(Session ses) {
throw DbException.getUnsupportedException("remove index");
}
/** {@inheritDoc} */
- @Override public void truncate(Session session) {
+ @Override public void truncate(Session ses) {
throw DbException.getUnsupportedException("truncate");
}
@@ -235,7 +253,7 @@ public abstract class GridMergeIndex extends BaseIndex {
}
/** {@inheritDoc} */
- @Override public Cursor findFirstOrLast(Session session, boolean first) {
+ @Override public Cursor findFirstOrLast(Session ses, boolean first) {
throw DbException.getUnsupportedException("findFirstOrLast");
}
@@ -299,6 +317,7 @@ public abstract class GridMergeIndex extends BaseIndex {
private Iterator<Row> stream;
/**
+ * @param stream Iterator.
*/
public FetchingCursor(Iterator<Row> stream) {
super(new FetchedIterator());
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 32d1c95..cde3288 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -33,6 +33,7 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
import org.apache.ignite.marshaller.*;
import org.apache.ignite.plugin.extensions.communication.*;
import org.h2.command.*;
@@ -47,6 +48,7 @@ import org.h2.table.*;
import org.h2.tools.*;
import org.h2.util.*;
import org.h2.value.*;
+import org.jetbrains.annotations.*;
import org.jsr166.*;
import javax.cache.*;
@@ -234,10 +236,15 @@ public class GridReduceQueryExecutor {
Object errState = r.state.get();
if (errState != null) {
+ CacheException err0 = errState instanceof CacheException ? (CacheException)errState : null;
+
+ if (err0 != null && err0.getCause() instanceof IgniteClientDisconnectedException)
+ throw err0;
+
CacheException e = new CacheException("Failed to fetch data from node: " + node.id());
- if (errState instanceof CacheException)
- e.addSuppressed((Throwable)errState);
+ if (err0 != null)
+ e.addSuppressed(err0);
throw e;
}
@@ -301,6 +308,7 @@ public class GridReduceQueryExecutor {
}
/**
+ * @param cctx Cache context.
* @return {@code true} If cache context
*/
private boolean hasMovingPartitions(GridCacheContext<?,?> cctx) {
@@ -481,6 +489,12 @@ public class GridReduceQueryExecutor {
runs.put(qryReqId, r);
try {
+ if (ctx.clientDisconnected()) {
+ throw new CacheException("Query was cancelled, client node disconnected.",
+ new IgniteClientDisconnectedException(ctx.cluster().clientReconnectFuture(),
+ "Client node disconnected."));
+ }
+
Collection<GridCacheSqlQuery> mapQrys = qry.mapQueries();
if (qry.explain()) {
@@ -506,8 +520,14 @@ public class GridReduceQueryExecutor {
Object state = r.state.get();
if (state != null) {
- if (state instanceof CacheException)
- throw new CacheException("Failed to run map query remotely.", (CacheException)state);
+ if (state instanceof CacheException) {
+ CacheException err = (CacheException)state;
+
+ if (err.getCause() instanceof IgniteClientDisconnectedException)
+ throw err;
+
+ throw new CacheException("Failed to run map query remotely.", err);
+ }
if (state instanceof AffinityTopologyVersion) {
retry = true;
@@ -550,7 +570,20 @@ public class GridReduceQueryExecutor {
catch (IgniteCheckedException | RuntimeException e) {
U.closeQuiet(r.conn);
- throw new CacheException("Failed to run reduce query locally.", e);
+ if (e instanceof CacheException)
+ throw (CacheException)e;
+
+ Throwable cause = e;
+
+ if (e instanceof IgniteCheckedException) {
+ Throwable disconnectedErr =
+ ((IgniteCheckedException)e).getCause(IgniteClientDisconnectedException.class);
+
+ if (disconnectedErr != null)
+ cause = disconnectedErr;
+ }
+
+ throw new CacheException("Failed to run reduce query locally.", cause);
}
finally {
if (!runs.remove(qryReqId, r))
@@ -1082,6 +1115,17 @@ public class GridReduceQueryExecutor {
}
/**
+ * @param reconnectFut Reconnect future.
+ */
+ public void onDisconnected(IgniteFuture<?> reconnectFut) {
+ CacheException err = new CacheException("Query was cancelled, client node disconnected.",
+ new IgniteClientDisconnectedException(reconnectFut, "Client node disconnected."));
+
+ for (Map.Entry<Long, QueryRun> e : runs.entrySet())
+ e.getValue().disconnected(err);
+ }
+
+ /**
*
*/
private static class QueryRun {
@@ -1104,7 +1148,7 @@ public class GridReduceQueryExecutor {
* @param o Fail state object.
* @param nodeId Node ID.
*/
- void state(Object o, UUID nodeId) {
+ void state(Object o, @Nullable UUID nodeId) {
assert o != null;
assert o instanceof CacheException || o instanceof AffinityTopologyVersion : o.getClass();
@@ -1117,6 +1161,20 @@ public class GridReduceQueryExecutor {
for (GridMergeTable tbl : tbls) // Fail all merge indexes.
tbl.getScanIndex(null).fail(nodeId);
}
+
+ /**
+ * @param e Error.
+ */
+ void disconnected(CacheException e) {
+ if (!state.compareAndSet(null, e))
+ return;
+
+ while (latch.getCount() != 0) // We don't need to wait for all nodes to reply.
+ latch.countDown();
+
+ for (GridMergeTable tbl : tbls) // Fail all merge indexes.
+ tbl.getScanIndex(null).fail(e);
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/57ac2b3b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientReconnectCacheQueriesFailoverTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientReconnectCacheQueriesFailoverTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientReconnectCacheQueriesFailoverTest.java
new file mode 100644
index 0000000..23320ae
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteClientReconnectCacheQueriesFailoverTest.java
@@ -0,0 +1,225 @@
+/*
+ * 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.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.cache.query.*;
+import org.apache.ignite.cache.query.annotations.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+
+import javax.cache.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.cache.CacheMode.*;
+
+/**
+ *
+ */
+public class IgniteClientReconnectCacheQueriesFailoverTest extends IgniteClientReconnectFailoverAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setIndexedTypes(Integer.class, Person.class);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ final IgniteCache<Integer, Person> cache = grid(serverCount()).cache(null);
+
+ assertNotNull(cache);
+
+ for (int i = 0; i <= 10_000; i++)
+ cache.put(i, new Person(i, "name-" + i));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectCacheQueries() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ final IgniteCache<Integer, Person> cache = client.cache(null);
+
+ assertNotNull(cache);
+
+ reconnectFailover(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ SqlQuery<Integer, Person> sqlQry = new SqlQuery<>(Person.class, "where id > 1");
+
+ try {
+ assertEquals(9999, cache.query(sqlQry).getAll().size());
+ }
+ catch (CacheException e) {
+ if (e.getCause() instanceof IgniteClientDisconnectedException)
+ throw e;
+ else
+ log.info("Ignore error: " + e);
+ }
+
+ try {
+ SqlFieldsQuery fieldsQry = new SqlFieldsQuery("select avg(p.id) from Person p");
+
+ List<List<?>> res = cache.query(fieldsQry).getAll();
+
+ assertEquals(1, res.size());
+
+ Double avg = (Double)res.get(0).get(0);
+
+ assertEquals(5_000, avg.intValue());
+ }
+ catch (CacheException e) {
+ if (e.getCause() instanceof IgniteClientDisconnectedException)
+ throw e;
+ else
+ log.info("Ignore error: " + e);
+ }
+
+ return null;
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testReconnectScanQuery() throws Exception {
+ final Ignite client = grid(serverCount());
+
+ final IgniteCache<Integer, Person> cache = client.cache(null);
+
+ assertNotNull(cache);
+
+ final Affinity<Integer> aff = client.affinity(null);
+
+ final Map<Integer, Integer> partMap = new HashMap<>();
+
+ for (int i = 0; i < aff.partitions(); i++)
+ partMap.put(i, 0);
+
+ for (int i = 0; i <= 10_000; i++) {
+ Integer part = aff.partition(i);
+
+ Integer size = partMap.get(part);
+
+ partMap.put(part, size + 1);
+ }
+
+ reconnectFailover(new Callable<Void>() {
+ @Override public Void call() throws Exception {
+ ScanQuery<Integer, Person> qry = new ScanQuery<>(new IgniteBiPredicate<Integer, Person>() {
+ @Override public boolean apply(Integer key, Person val) {
+ return val.getId() % 2 == 1;
+ }
+ });
+
+ assertEquals(5000, cache.query(qry).getAll().size());
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ Integer part = rnd.nextInt(0, aff.partitions());
+
+ qry = new ScanQuery<>(part);
+
+ assertEquals((int)partMap.get(part), cache.query(qry).getAll().size());
+
+ return null;
+ }
+ });
+ }
+
+ /**
+ *
+ */
+ public static class Person {
+ /** */
+ @QuerySqlField
+ public int id;
+
+ /** */
+ @QuerySqlField
+ public String name;
+
+ /**
+ * @param id Id.
+ * @param name Name.
+ */
+ public Person(int id, String name) {
+ this.id = id;
+ this.name = name;
+ }
+
+ /**
+ * @return Id.
+ */
+ public int getId() {
+ return id;
+ }
+
+ /**
+ * @param id Set id.
+ */
+ public void setId(int id) {
+ this.id = id;
+ }
+
+ /**
+ * @return Name.
+ */
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * @param name Name.
+ */
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ return this == o || !(o == null || getClass() != o.getClass()) && id == ((Person)o).id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(Person.class, this);
+ }
+ }
+}