http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java index 63db7a2..5ce2efd 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java @@ -28,9 +28,11 @@ import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.lang.*; import org.apache.ignite.marshaller.optimized.*; import org.apache.ignite.resources.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; import org.apache.ignite.testframework.junits.common.*; -import java.io.*; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.*; @@ -42,7 +44,7 @@ import static org.apache.ignite.internal.processors.continuous.GridContinuousPro /** * Event consume test. */ -public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements Serializable { +public class GridEventConsumeSelfTest extends GridCommonAbstractTest { /** */ private static final String PRJ_PRED_CLS_NAME = "org.apache.ignite.tests.p2p.GridEventConsumeProjectionPredicate"; @@ -55,6 +57,9 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements /** Number of created consumes per thread in multithreaded test. */ private static final int CONSUME_CNT = 500; + /** */ + private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true); + /** Consume latch. */ private static volatile CountDownLatch consumeLatch; @@ -71,6 +76,12 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { IgniteConfiguration cfg = super.getConfiguration(gridName); + TcpDiscoverySpi disc = new TcpDiscoverySpi(); + + disc.setIpFinder(IP_FINDER); + + cfg.setDiscoverySpi(disc); + if (include) cfg.setUserAttributes(F.asMap("include", true)); @@ -78,7 +89,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements } /** {@inheritDoc} */ - @Override protected void beforeTestsStarted() throws Exception { + @Override protected void beforeTest() throws Exception { assertTrue(GRID_CNT > 1); include = true; @@ -91,32 +102,33 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements } /** {@inheritDoc} */ - @Override protected void afterTestsStopped() throws Exception { - stopAllGrids(); - } - - /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { - assertEquals(GRID_CNT, grid(0).cluster().nodes().size()); + try { + assertEquals(GRID_CNT, grid(0).cluster().nodes().size()); - for (int i = 0; i < GRID_CNT; i++) { - IgniteKernal grid = (IgniteKernal)grid(i); + for (int i = 0; i < GRID_CNT; i++) { + IgniteEx grid = grid(i); - GridContinuousProcessor proc = grid.context().continuous(); + GridContinuousProcessor proc = grid.context().continuous(); - if (noAutoUnsubscribe) { - localRoutines(proc).clear(); + try { + if (!noAutoUnsubscribe) { + Map rmtInfos = U.field(proc, "rmtInfos"); - U.<Map>field(proc, "rmtInfos").clear(); - } + assertTrue("Unexpected remote infos: " + rmtInfos, rmtInfos.isEmpty()); + } + } + finally { + U.<Map>field(proc, "rmtInfos").clear(); + } - assertEquals(0, localRoutines(proc).size()); - assertEquals(0, U.<Map>field(proc, "rmtInfos").size()); - assertEquals(0, U.<Map>field(proc, "startFuts").size()); - assertEquals(0, U.<Map>field(proc, "waitForStartAck").size()); - assertEquals(0, U.<Map>field(proc, "stopFuts").size()); - assertEquals(0, U.<Map>field(proc, "waitForStopAck").size()); - assertEquals(0, U.<Map>field(proc, "pending").size()); + assertEquals(0, U.<Map>field(proc, "rmtInfos").size()); + assertEquals(0, U.<Map>field(proc, "startFuts").size()); + assertEquals(0, U.<Map>field(proc, "stopFuts").size()); + } + } + finally { + stopAllGrids(); } } @@ -509,28 +521,6 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements /** * @throws Exception If failed. */ - public void testEmptyProjection() throws Exception { - try { - events(grid(0).cluster().forPredicate(F.<ClusterNode>alwaysFalse())).remoteListen( - new P2<UUID, Event>() { - @Override public boolean apply(UUID nodeId, Event evt) { - return true; - } - }, - null - ); - - assert false : "Exception was not thrown."; - } - catch (IgniteException e) { - assertTrue(e.getMessage().startsWith( - "Failed to register remote continuous listener (projection is empty).")); - } - } - - /** - * @throws Exception If failed. - */ public void testStopByCallback() throws Exception { final Collection<UUID> nodeIds = new HashSet<>(); final AtomicInteger cnt = new AtomicInteger(); @@ -663,7 +653,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements final CountDownLatch latch = new CountDownLatch(GRID_CNT + 1); UUID consumeId = grid(0).events().remoteListen( - new P2<UUID, Event>() { + notSerializableProxy(new P2<UUID, Event>() { @Override public boolean apply(UUID nodeId, Event evt) { info("Event from " + nodeId + " [" + evt.shortDisplay() + ']'); @@ -675,12 +665,12 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements return true; } - }, - new P1<Event>() { + }), + notSerializableProxy(new P1<Event>() { @Override public boolean apply(Event evt) { return evt.type() == EVT_JOB_STARTED; } - }, + }), EVT_JOB_STARTED, EVT_JOB_FINISHED ); @@ -871,14 +861,14 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements final CountDownLatch latch = new CountDownLatch(GRID_CNT); for (int i = 0; i < GRID_CNT; i++) { - grid(0).events().localListen(new IgnitePredicate<Event>() { + grid(i).events().localListen(new IgnitePredicate<Event>() { @Override public boolean apply(Event evt) { if (nodeId.equals(((DiscoveryEvent) evt).eventNode().id())) latch.countDown(); return true; } - }, EVT_NODE_LEFT); + }, EVT_NODE_LEFT, EVT_NODE_FAILED); } g.events().remoteListen( @@ -893,7 +883,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements stopGrid("anotherGrid"); - latch.await(); + assert latch.await(3000, MILLISECONDS); } /** @@ -939,7 +929,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest implements stopGrid("anotherGrid"); - discoLatch.await(); + discoLatch.await(3000, MILLISECONDS); grid(0).compute().broadcast(F.noop());
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java index 280ba95..80bf9e0 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessorSelfTest.java @@ -84,6 +84,7 @@ public class DataStreamProcessorSelfTest extends GridCommonAbstractTest { TcpDiscoverySpi spi = new TcpDiscoverySpi(); + spi.setForceServerMode(true); spi.setIpFinder(ipFinder); cfg.setDiscoverySpi(spi); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultiThreadedSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultiThreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultiThreadedSelfTest.java index bea3809..e0092d4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultiThreadedSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultiThreadedSelfTest.java @@ -41,6 +41,9 @@ public class DataStreamerMultiThreadedSelfTest extends GridCommonAbstractTest { /** IP finder. */ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true); + /** */ + private boolean dynamicCache; + /** {@inheritDoc} */ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { IgniteConfiguration cfg = super.getConfiguration(gridName); @@ -50,13 +53,22 @@ public class DataStreamerMultiThreadedSelfTest extends GridCommonAbstractTest { cfg.setDiscoverySpi(discoSpi); + if (!dynamicCache) + cfg.setCacheConfiguration(cacheConfiguration()); + + return cfg; + } + + /** + * @return Cache configuration. + */ + private CacheConfiguration cacheConfiguration() { CacheConfiguration ccfg = defaultCacheConfiguration(); ccfg.setCacheMode(PARTITIONED); ccfg.setBackups(1); - cfg.setCacheConfiguration(ccfg); - return cfg; + return ccfg; } /** {@inheritDoc} */ @@ -68,8 +80,22 @@ public class DataStreamerMultiThreadedSelfTest extends GridCommonAbstractTest { * @throws Exception If failed. */ public void testStartStopIgnites() throws Exception { - fail("https://issues.apache.org/jira/browse/IGNITE-840";); + startStopIgnites(); + } + + /** + * @throws Exception If failed. + */ + public void testStartStopIgnitesDynamicCache() throws Exception { + dynamicCache = true; + + startStopIgnites(); + } + /** + * @throws Exception If failed. + */ + private void startStopIgnites() throws Exception { for (int attempt = 0; attempt < 3; ++attempt) { log.info("Iteration: " + attempt); @@ -77,28 +103,29 @@ public class DataStreamerMultiThreadedSelfTest extends GridCommonAbstractTest { Set<IgniteFuture> futs = new HashSet<>(); - IgniteInternalFuture<?> fut; + final AtomicInteger igniteId = new AtomicInteger(1); - try (final DataStreamerImpl dataLdr = (DataStreamerImpl)ignite.dataStreamer(null)) { - dataLdr.maxRemapCount(0); + IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + for (int i = 1; i < 5; ++i) + startGrid(igniteId.incrementAndGet()); - final AtomicInteger igniteId = new AtomicInteger(1); + return true; + } + }, 2, "start-node-thread"); - fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() { - @Override public Object call() throws Exception { - for (int i = 1; i < 5; ++i) - startGrid(igniteId.incrementAndGet()); + if (dynamicCache) + ignite.getOrCreateCache(cacheConfiguration()); - return true; - } - }, 2, "start-node-thread"); + try (final DataStreamerImpl dataLdr = (DataStreamerImpl)ignite.dataStreamer(null)) { + dataLdr.maxRemapCount(0); - Random random = new Random(); + Random rnd = new Random(); long endTime = U.currentTimeMillis() + 15_000; while (!fut.isDone() && U.currentTimeMillis() < endTime) - futs.add(dataLdr.addData(random.nextInt(100_000), random.nextInt(100_000))); + futs.add(dataLdr.addData(rnd.nextInt(100_000), String.valueOf(rnd.nextInt(100_000)))); } for (IgniteFuture f : futs) http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultinodeCreateCacheTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultinodeCreateCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultinodeCreateCacheTest.java new file mode 100644 index 0000000..12b6458 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerMultinodeCreateCacheTest.java @@ -0,0 +1,97 @@ +/* + * 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.datastreamer; + +import org.apache.ignite.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.internal.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.testframework.*; +import org.apache.ignite.testframework.junits.common.*; + +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +/** + * + */ +public class DataStreamerMultinodeCreateCacheTest extends GridCommonAbstractTest { + /** */ + private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true); + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(IP_FINDER); + + ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setSocketTimeout(50); + ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setAckTimeout(50); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testCreateCacheAndStream() throws Exception { + final int THREADS = 5; + + startGrids(THREADS); + + final AtomicInteger idx = new AtomicInteger(); + + IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + int threadIdx = idx.getAndIncrement(); + + long stopTime = System.currentTimeMillis() + 60_000; + + Ignite ignite = grid(threadIdx); + + int iter = 0; + + while (System.currentTimeMillis() < stopTime) { + String cacheName = "cache-" + threadIdx + "-" + (iter % 10); + + try (IgniteCache<Integer, String> cache = ignite.getOrCreateCache(cacheName)) { + try (IgniteDataStreamer<Object, Object> stmr = ignite.dataStreamer(cacheName)) { + ((DataStreamerImpl<Object, Object>)stmr).maxRemapCount(0); + + for (int i = 0; i < 1000; i++) + stmr.addData(i, i); + } + } + + iter++; + } + + return null; + } + }, THREADS, "create-cache"); + + fut.get(2 * 60_000); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsClientCacheSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsClientCacheSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsClientCacheSelfTest.java index d983302..9cda1b4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsClientCacheSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsClientCacheSelfTest.java @@ -85,13 +85,16 @@ public class IgfsClientCacheSelfTest extends IgfsAbstractSelfTest { cfg.setCacheConfiguration(cacheConfiguration(META_CACHE_NAME), cacheConfiguration(DATA_CACHE_NAME), cacheConfiguration(CACHE_NAME)); - if (!gridName.equals(getTestGridName(0))) - cfg.setClientMode(true); - TcpDiscoverySpi disco = new TcpDiscoverySpi(); disco.setIpFinder(IP_FINDER); + if (!gridName.equals(getTestGridName(0))) { + cfg.setClientMode(true); + + disco.setForceServerMode(true); + } + cfg.setDiscoverySpi(disco); FileSystemConfiguration igfsCfg = new FileSystemConfiguration(); @@ -114,11 +117,12 @@ public class IgfsClientCacheSelfTest extends IgfsAbstractSelfTest { cacheCfg.setName(cacheName); + cacheCfg.setNearConfiguration(null); + if (META_CACHE_NAME.equals(cacheName)) cacheCfg.setCacheMode(REPLICATED); else { cacheCfg.setCacheMode(PARTITIONED); - cacheCfg.setNearConfiguration(null); cacheCfg.setBackups(0); cacheCfg.setAffinityMapper(new IgfsGroupDataBlocksKeyMapper(128)); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsCommonAbstractTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsCommonAbstractTest.java index fa7f048..e879130 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsCommonAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsCommonAbstractTest.java @@ -46,16 +46,6 @@ public class IgfsCommonAbstractTest extends GridCommonAbstractTest { } /** {@inheritDoc} */ - @Override protected IgniteConfiguration getConfiguration(IgniteTestResources rsrcs) throws Exception { - IgniteConfiguration cfg = super.getConfiguration(rsrcs); - - cfg.setPeerClassLoadingEnabled(false); - cfg.setLocalHost("127.0.0.1"); - - return cfg; - } - - /** {@inheritDoc} */ @Override protected IgniteConfiguration getConfiguration(String gridName, IgniteTestResources rsrcs) throws Exception { IgniteConfiguration cfg = super.getConfiguration(gridName, rsrcs); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsModesSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsModesSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsModesSelfTest.java index 3c90af2..5b80044 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsModesSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsModesSelfTest.java @@ -471,7 +471,9 @@ public class IgfsModesSelfTest extends IgfsCommonAbstractTest { * * @throws Exception If failed. */ - public void _testPropagationDualAsync() throws Exception { + public void testPropagationDualAsync() throws Exception { + fail("https://issues.apache.org/jira/browse/IGNITE-822";); + mode = DUAL_ASYNC; checkPropagation(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsOneClientNodeTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsOneClientNodeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsOneClientNodeTest.java index 49ddb03..3498cd9 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsOneClientNodeTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/igfs/IgfsOneClientNodeTest.java @@ -52,11 +52,9 @@ public class IgfsOneClientNodeTest extends GridCommonAbstractTest { cfg.setClientMode(true); - TcpDiscoverySpi discoSpi = new TcpDiscoverySpi(); - - discoSpi.setIpFinder(new TcpDiscoveryVmIpFinder(true)); - - cfg.setDiscoverySpi(discoSpi); + cfg.setDiscoverySpi(new TcpDiscoverySpi() + .setForceServerMode(true) + .setIpFinder(new TcpDiscoveryVmIpFinder(true))); FileSystemConfiguration igfsCfg = new FileSystemConfiguration(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ClosureServiceClientsNodesTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ClosureServiceClientsNodesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ClosureServiceClientsNodesTest.java index 761f00f..faccc9a 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ClosureServiceClientsNodesTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/ClosureServiceClientsNodesTest.java @@ -53,11 +53,7 @@ public class ClosureServiceClientsNodesTest extends GridCommonAbstractTest { cfg.setMarshaller(new OptimizedMarshaller(false)); - TcpDiscoverySpi discoSpi = new TcpDiscoverySpi(); - - discoSpi.setIpFinder(ipFinder); - - cfg.setDiscoverySpi(discoSpi); + cfg.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(ipFinder).setForceServerMode(true)); cfg.setCacheConfiguration(); @@ -88,6 +84,8 @@ public class ClosureServiceClientsNodesTest extends GridCommonAbstractTest { srvNames.add(getTestGridName(i)); for (int i = 0 ; i < NODES_CNT; i++) { + log.info("Iteration: " + i); + Ignite ignite = grid(i); Collection<String> res = ignite.compute().broadcast(new IgniteCallable<String>() { @@ -113,6 +111,8 @@ public class ClosureServiceClientsNodesTest extends GridCommonAbstractTest { */ public void testClientClosure() throws Exception { for (int i = 0 ; i < NODES_CNT; i++) { + log.info("Iteration: " + i); + Ignite ignite = grid(i); Collection<String> res = ignite.compute(ignite.cluster().forClients()). @@ -138,6 +138,8 @@ public class ClosureServiceClientsNodesTest extends GridCommonAbstractTest { */ public void testCustomClosure() throws Exception { for (int i = 0 ; i < NODES_CNT; i++) { + log.info("Iteration: " + i); + Ignite ignite = grid(i); Collection<String> res = ignite.compute(ignite.cluster().forPredicate(F.<ClusterNode>alwaysTrue())). @@ -161,6 +163,8 @@ public class ClosureServiceClientsNodesTest extends GridCommonAbstractTest { UUID clientNodeId = grid(0).cluster().localNode().id(); for (int i = 0 ; i < NODES_CNT; i++) { + log.info("Iteration: " + i); + Ignite ignite = grid(i); ignite.services().deployNodeSingleton(SINGLETON_NAME, new TestService()); @@ -194,6 +198,8 @@ public class ClosureServiceClientsNodesTest extends GridCommonAbstractTest { UUID clientNodeId = grid(0).cluster().localNode().id(); for (int i = 0 ; i < NODES_CNT; i++) { + log.info("Iteration: " + i); + Ignite ignite = grid(i); ignite.services(ignite.cluster().forClients()).deployNodeSingleton(SINGLETON_NAME, new TestService()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceClientNodeTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceClientNodeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceClientNodeTest.java new file mode 100644 index 0000000..404c32b --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/service/GridServiceClientNodeTest.java @@ -0,0 +1,81 @@ +/* + * 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.service; + +import org.apache.ignite.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.testframework.junits.common.*; + +import java.util.concurrent.*; + +/** + * + */ +public class GridServiceClientNodeTest extends GridCommonAbstractTest { + /** */ + protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** */ + private static final int NODE_CNT = 3; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + ((TcpDiscoverySpi) cfg.getDiscoverySpi()).setIpFinder(ipFinder); + + if (gridName.equals(getTestGridName(NODE_CNT - 1))) + cfg.setClientMode(true); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + super.beforeTestsStarted(); + + startGrids(NODE_CNT); + } + + /** {@inheritDoc} */ + @Override protected void afterTestsStopped() throws Exception { + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testDeployFromClient() throws Exception { + Ignite ignite = ignite(NODE_CNT - 1); + + assertTrue(ignite.configuration().isClientMode()); + + String svcName = "testService"; + + CountDownLatch latch = new CountDownLatch(1); + + DummyService.exeLatch(svcName, latch); + + ignite.services().deployClusterSingleton(svcName, new DummyService()); + + assertTrue(latch.await(5000, TimeUnit.MILLISECONDS)); + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferSelfTest.java new file mode 100644 index 0000000..cbf7d89 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferSelfTest.java @@ -0,0 +1,112 @@ +/* + * 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.util.nio; + +import junit.framework.TestCase; + +import java.nio.*; +import java.util.*; + +/** + * Tests for {@link GridNioDelimitedBuffer}. + */ +public class GridNioDelimitedBufferSelfTest extends TestCase { + /** */ + private static final String ASCII = "ASCII"; + + /** + * Tests simple delimiter (excluded from alphabet) + */ + public void testReadZString() throws Exception { + Random rnd = new Random(); + + int buffSize = 0; + + byte[] delim = new byte[] {0}; + + List<String> strs = new ArrayList<>(50); + + for (int i = 0; i < 50; i++) { + int len = rnd.nextInt(128) + 1; + + buffSize += len + delim.length; + + StringBuilder sb = new StringBuilder(len); + + for (int j = 0; j < len; j++) + sb.append((char)(rnd.nextInt(26) + 'a')); + + + strs.add(sb.toString()); + } + + ByteBuffer buff = ByteBuffer.allocate(buffSize); + + for (String str : strs) { + buff.put(str.getBytes(ASCII)); + buff.put(delim); + } + + buff.flip(); + + byte[] msg; + + GridNioDelimitedBuffer delimBuff = new GridNioDelimitedBuffer(delim); + + List<String> res = new ArrayList<>(strs.size()); + + while ((msg = delimBuff.read(buff)) != null) + res.add(new String(msg, ASCII)); + + assertEquals(strs, res); + } + + /** + * Tests compound delimiter (included to alphabet) + */ + public void testDelim() throws Exception { + byte[] delim = "aabb".getBytes(ASCII); + + List<String> strs = Arrays.asList("za", "zaa", "zaab", "zab", "zaabaababbbbabaab"); + + int buffSize = 0; + + for (String str : strs) + buffSize += str.length() + delim.length; + + ByteBuffer buff = ByteBuffer.allocate(buffSize); + + for (String str : strs) { + buff.put(str.getBytes(ASCII)); + buff.put(delim); + } + + buff.flip(); + + byte[] msg; + + GridNioDelimitedBuffer delimBuff = new GridNioDelimitedBuffer(delim); + + List<String> res = new ArrayList<>(strs.size()); + + while ((msg = delimBuff.read(buff)) != null) + res.add(new String(msg, ASCII)); + + assertEquals(strs, res); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java deleted file mode 100644 index a0dd2e5..0000000 --- a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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.util.nio; - -import junit.framework.TestCase; - -import java.nio.*; -import java.util.*; - -/** - * Tests for {@link GridNioDelimitedBuffer}. - */ -public class GridNioDelimitedBufferTest extends TestCase { - /** */ - private static final String ASCII = "ASCII"; - - /** - * Tests simple delimiter (excluded from alphabet) - */ - public void testReadZString() throws Exception { - Random rnd = new Random(); - - int buffSize = 0; - - byte[] delim = new byte[] {0}; - - List<String> strs = new ArrayList<>(50); - - for (int i = 0; i < 50; i++) { - int len = rnd.nextInt(128) + 1; - - buffSize += len + delim.length; - - StringBuilder sb = new StringBuilder(len); - - for (int j = 0; j < len; j++) - sb.append((char)(rnd.nextInt(26) + 'a')); - - - strs.add(sb.toString()); - } - - ByteBuffer buff = ByteBuffer.allocate(buffSize); - - for (String str : strs) { - buff.put(str.getBytes(ASCII)); - buff.put(delim); - } - - buff.flip(); - - byte[] msg; - - GridNioDelimitedBuffer delimBuff = new GridNioDelimitedBuffer(delim); - - List<String> res = new ArrayList<>(strs.size()); - - while ((msg = delimBuff.read(buff)) != null) - res.add(new String(msg, ASCII)); - - assertEquals(strs, res); - } - - /** - * Tests compound delimiter (included to alphabet) - */ - public void testDelim() throws Exception { - byte[] delim = "aabb".getBytes(ASCII); - - List<String> strs = Arrays.asList("za", "zaa", "zaab", "zab", "zaabaababbbbabaab"); - - int buffSize = 0; - - for (String str : strs) - buffSize += str.length() + delim.length; - - ByteBuffer buff = ByteBuffer.allocate(buffSize); - - for (String str : strs) { - buff.put(str.getBytes(ASCII)); - buff.put(delim); - } - - buff.flip(); - - byte[] msg; - - GridNioDelimitedBuffer delimBuff = new GridNioDelimitedBuffer(delim); - - List<String> res = new ArrayList<>(strs.size()); - - while ((msg = delimBuff.read(buff)) != null) - res.add(new String(msg, ASCII)); - - assertEquals(strs, res); - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java index 63521e8..c81ed56 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSelfTest.java @@ -1286,7 +1286,7 @@ public class GridNioSelfTest extends GridCommonAbstractTest { } /** - * Test client to use instead of {@link GridTcpCommunicationClient} + * Test client to use instead of {@link GridTcpNioCommunicationClient} */ private static class TestClient implements AutoCloseable { /** Socket implementation to use. */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/internal/util/offheap/unsafe/GridUnsafeMemorySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/offheap/unsafe/GridUnsafeMemorySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/offheap/unsafe/GridUnsafeMemorySelfTest.java index 00f89f9..51199e6 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/util/offheap/unsafe/GridUnsafeMemorySelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/offheap/unsafe/GridUnsafeMemorySelfTest.java @@ -257,7 +257,9 @@ public class GridUnsafeMemorySelfTest extends GridCommonAbstractTest { /** * @throws Exception If failed. */ - public void _testGuardedOpsPerformance() throws Exception { + public void testGuardedOpsPerformance() throws Exception { + fail("https://issues.apache.org/jira/browse/IGNITE-823";); + final GridUnsafeGuard guard = new GridUnsafeGuard(); final AtomicInteger i = new AtomicInteger(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/loadtests/GridCacheMultiNodeLoadTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/GridCacheMultiNodeLoadTest.java b/modules/core/src/test/java/org/apache/ignite/loadtests/GridCacheMultiNodeLoadTest.java index d106f5b..0030ce1 100644 --- a/modules/core/src/test/java/org/apache/ignite/loadtests/GridCacheMultiNodeLoadTest.java +++ b/modules/core/src/test/java/org/apache/ignite/loadtests/GridCacheMultiNodeLoadTest.java @@ -65,7 +65,10 @@ public class GridCacheMultiNodeLoadTest extends GridCommonAbstractTest { cacheCfg.setStartSize(10); cacheCfg.setWriteSynchronizationMode(FULL_SYNC); - cacheCfg.setEvictionPolicy(new LruEvictionPolicy(100000)); + LruEvictionPolicy plc = new LruEvictionPolicy(); + plc.setMaxSize(100000); + + cacheCfg.setEvictionPolicy(plc); cacheCfg.setBackups(1); cacheCfg.setRebalanceMode(SYNC); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/loadtests/datastructures/GridCachePartitionedAtomicLongLoadTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/datastructures/GridCachePartitionedAtomicLongLoadTest.java b/modules/core/src/test/java/org/apache/ignite/loadtests/datastructures/GridCachePartitionedAtomicLongLoadTest.java index 6960fa1..0d9ec8f 100644 --- a/modules/core/src/test/java/org/apache/ignite/loadtests/datastructures/GridCachePartitionedAtomicLongLoadTest.java +++ b/modules/core/src/test/java/org/apache/ignite/loadtests/datastructures/GridCachePartitionedAtomicLongLoadTest.java @@ -70,7 +70,11 @@ public class GridCachePartitionedAtomicLongLoadTest extends GridCommonAbstractTe cc.setStartSize(200); cc.setRebalanceMode(CacheRebalanceMode.SYNC); cc.setWriteSynchronizationMode(FULL_SYNC); - cc.setEvictionPolicy(new LruEvictionPolicy<>(1000)); + + LruEvictionPolicy plc = new LruEvictionPolicy(); + plc.setMaxSize(1000); + + cc.setEvictionPolicy(plc); cc.setBackups(1); cc.setAffinity(new RendezvousAffinityFunction(true)); cc.setEvictSynchronized(true); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/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 0b0f099..77d3905 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 @@ -54,11 +54,13 @@ public class GridCacheTestContext<K, V> extends GridCacheContext<K, V> { new GridCacheMvccManager(), new GridCacheDeploymentManager<K, V>(), new GridCachePartitionExchangeManager<K, V>(), - new GridCacheIoManager() + new GridCacheIoManager(), + null ), defaultCacheConfiguration(), CacheType.USER, true, + true, new GridCacheEventManager(), new GridCacheSwapManager(false), new CacheOsStoreManager(null, new CacheConfiguration()), http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/loadtests/swap/GridSwapEvictAllBenchmark.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/loadtests/swap/GridSwapEvictAllBenchmark.java b/modules/core/src/test/java/org/apache/ignite/loadtests/swap/GridSwapEvictAllBenchmark.java index 1561b77..62066da 100644 --- a/modules/core/src/test/java/org/apache/ignite/loadtests/swap/GridSwapEvictAllBenchmark.java +++ b/modules/core/src/test/java/org/apache/ignite/loadtests/swap/GridSwapEvictAllBenchmark.java @@ -266,7 +266,11 @@ public class GridSwapEvictAllBenchmark { ccfg.setSwapEnabled(true); ccfg.setEvictSynchronized(false); - ccfg.setEvictionPolicy(new FifoEvictionPolicy(EVICT_PLC_SIZE)); + + FifoEvictionPolicy plc = new FifoEvictionPolicy(); + plc.setMaxSize(EVICT_PLC_SIZE); + + ccfg.setEvictionPolicy(plc); if (store != null) { ccfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store)); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedMarshallerNodeFailoverTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedMarshallerNodeFailoverTest.java b/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedMarshallerNodeFailoverTest.java index 07fd9e3..35abf7e 100644 --- a/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedMarshallerNodeFailoverTest.java +++ b/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedMarshallerNodeFailoverTest.java @@ -113,7 +113,7 @@ public class OptimizedMarshallerNodeFailoverTest extends GridCommonAbstractTest cache = true; - Ignite ignite = startGrid(2); // Check can start on more cache node. + Ignite ignite = startGrid(2); // Check can start one more cache node. assertNotNull(ignite.cache(null)); } @@ -122,7 +122,7 @@ public class OptimizedMarshallerNodeFailoverTest extends GridCommonAbstractTest * @throws Exception If failed. */ public void testRestartAllNodes() throws Exception { - cache = false; + cache = true; String home = U.getIgniteHome(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingNoPeerClassLoadingSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingNoPeerClassLoadingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingNoPeerClassLoadingSelfTest.java index b6a775f..37d2ec7 100644 --- a/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingNoPeerClassLoadingSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingNoPeerClassLoadingSelfTest.java @@ -17,10 +17,8 @@ package org.apache.ignite.messaging; -import org.apache.ignite.*; import org.apache.ignite.configuration.*; import org.apache.ignite.internal.util.typedef.*; -import org.apache.ignite.resources.*; import org.apache.ignite.testframework.config.*; import java.net.*; @@ -33,6 +31,9 @@ import java.util.concurrent.atomic.*; * peer class loading. */ public class GridMessagingNoPeerClassLoadingSelfTest extends GridMessagingSelfTest { + /** */ + private static CountDownLatch rcvLatch; + /** {@inheritDoc} */ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { IgniteConfiguration cfg = super.getConfiguration(gridName); @@ -56,61 +57,37 @@ public class GridMessagingNoPeerClassLoadingSelfTest extends GridMessagingSelfTe Class rcCls = extLdr.loadClass(EXT_RESOURCE_CLS_NAME); - MessageListener list = new MessageListener(ignite1); - - ignite2.message().remoteListen("", list); - - message(ignite1.cluster().forRemotes()).send(null, Collections.singleton(rcCls.newInstance())); - - /* - We shouldn't get a message, because remote node won't be able to - unmarshal it (peer class loading is disabled.) - */ - assertFalse(list.rcvLatch.await(3, TimeUnit.SECONDS)); - } - - /** - * - */ - private static class MessageListener<UUID, Object> implements P2<UUID, Object> { - /** */ final AtomicBoolean error = new AtomicBoolean(false); //to make it modifiable - /** */ - final CountDownLatch rcvLatch = new CountDownLatch(1); + rcvLatch = new CountDownLatch(1); - /** */ - final Ignite sender; + ignite2.message().remoteListen(null, new P2<UUID, Object>() { + @Override public boolean apply(UUID nodeId, Object msg) { + try { + log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); - /** */ - @LoggerResource - private transient IgniteLogger log; + if (!nodeId.equals(ignite1.cluster().localNode().id())) { + log.error("Unexpected sender node: " + nodeId); - /** - * @param sender - */ - private MessageListener(Ignite sender) { - this.sender = sender; - } - - /** {@inheritDoc} */ - @Override public boolean apply(UUID nodeId, Object msg) { - try { - log.info("Received new message [msg=" + msg + ", senderNodeId=" + nodeId + ']'); + error.set(true); - if (!nodeId.equals(sender.cluster().localNode().id())) { - log.error("Unexpected sender node: " + nodeId); + return false; + } - error.set(true); - - return false; + return true; + } + finally { + rcvLatch.countDown(); } - - return true; - } - finally { - rcvLatch.countDown(); } - } + }); + + message(ignite1.cluster().forRemotes()).send(null, Collections.singleton(rcCls.newInstance())); + + /* + We shouldn't get a message, because remote node won't be able to + unmarshal it (peer class loading is disabled.) + */ + assertFalse(rcvLatch.await(3, TimeUnit.SECONDS)); } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingSelfTest.java index 2090d36..c11aeb0 100644 --- a/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/messaging/GridMessagingSelfTest.java @@ -44,7 +44,7 @@ import static org.apache.ignite.testframework.GridTestUtils.*; /** * Various tests for Messaging public API. */ -public class GridMessagingSelfTest extends GridCommonAbstractTest { +public class GridMessagingSelfTest extends GridCommonAbstractTest implements Serializable { /** */ private static final String MSG_1 = "MSG-1"; @@ -73,7 +73,7 @@ public class GridMessagingSelfTest extends GridCommonAbstractTest { public static final String EXT_RESOURCE_CLS_NAME = "org.apache.ignite.tests.p2p.TestUserResource"; /** Shared IP finder. */ - private final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + private final transient TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); /** * A test message topic. http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/messaging/IgniteMessagingWithClientTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/messaging/IgniteMessagingWithClientTest.java b/modules/core/src/test/java/org/apache/ignite/messaging/IgniteMessagingWithClientTest.java new file mode 100644 index 0000000..09abcdb --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/messaging/IgniteMessagingWithClientTest.java @@ -0,0 +1,166 @@ +/* + * 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.messaging; + +import org.apache.ignite.*; +import org.apache.ignite.cluster.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.internal.*; +import org.apache.ignite.internal.util.typedef.internal.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.marshaller.optimized.*; +import org.apache.ignite.resources.*; +import org.apache.ignite.spi.discovery.tcp.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.testframework.*; +import org.apache.ignite.testframework.junits.common.*; + +import java.io.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; + +/** + * + */ +public class IgniteMessagingWithClientTest extends GridCommonAbstractTest implements Serializable { + /** */ + protected static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** Message topic. */ + private enum TOPIC { + /** */ + ORDERED + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + cfg.setMarshaller(new OptimizedMarshaller(false)); + + if (gridName.equals(getTestGridName(2))) { + cfg.setClientMode(true); + + ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true); + } + + ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testMessageSendWithClientJoin() throws Exception { + fail("https://issues.apache.org/jira/browse/IGNITE-996";); + + startGrid(0); + + Ignite ignite1 = startGrid(1); + + ClusterGroup rmts = ignite1.cluster().forRemotes(); + + IgniteMessaging msg = ignite1.message(rmts); + + msg.localListen(TOPIC.ORDERED, new LocalListener()); + + msg.remoteListen(TOPIC.ORDERED, new RemoteListener()); + + final AtomicBoolean stop = new AtomicBoolean(); + + IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + int iter = 0; + + while (!stop.get()) { + if (iter % 10 == 0) + log.info("Client start/stop iteration: " + iter); + + iter++; + + try (Ignite ignite = startGrid(2)) { + assertTrue(ignite.configuration().isClientMode()); + } + } + + return null; + } + }, 1, "client-start-stop"); + + try { + long stopTime = U.currentTimeMillis() + 30_000; + + int iter = 0; + + while (System.currentTimeMillis() < stopTime) { + try { + ignite1.message(rmts).sendOrdered(TOPIC.ORDERED, Integer.toString(iter), 0); + } + catch (IgniteException e) { + log.info("Message send failed: " + e); + } + + iter++; + + if (iter % 100 == 0) + Thread.sleep(5); + } + } + finally { + stop.set(true); + } + + fut.get(); + } + + /** + * + */ + private static class LocalListener implements IgniteBiPredicate<UUID, String> { + /** {@inheritDoc} */ + @Override public boolean apply(UUID uuid, String s) { + return true; + } + } + + /** + * + */ + private static class RemoteListener implements IgniteBiPredicate<UUID, String> { + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** {@inheritDoc} */ + @Override public boolean apply(UUID nodeId, String msg) { + ignite.message(ignite.cluster().forNodeId(nodeId)).send(TOPIC.ORDERED, msg); + + return true; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java index ea51aff..8d27485 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java @@ -64,7 +64,7 @@ abstract class GridTcpCommunicationSpiAbstractTest extends GridAbstractCommunica // Test idle clients remove. for (CommunicationSpi spi : spis.values()) { - ConcurrentMap<UUID, GridTcpCommunicationClient> clients = U.field(spi, "clients"); + ConcurrentMap<UUID, GridCommunicationClient> clients = U.field(spi, "clients"); assertEquals(2, clients.size()); @@ -77,7 +77,7 @@ abstract class GridTcpCommunicationSpiAbstractTest extends GridAbstractCommunica super.afterTest(); for (CommunicationSpi spi : spis.values()) { - ConcurrentMap<UUID, GridTcpCommunicationClient> clients = U.field(spi, "clients"); + ConcurrentMap<UUID, GridCommunicationClient> clients = U.field(spi, "clients"); for (int i = 0; i < 20 && !clients.isEmpty(); i++) { info("Check failed for SPI [grid=" + http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java index c038ee7..2d175f5 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java @@ -256,7 +256,7 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic assertTrue(latch.await(10, TimeUnit.SECONDS)); for (CommunicationSpi spi : spis) { - ConcurrentMap<UUID, GridTcpCommunicationClient> clients = U.field(spi, "clients"); + ConcurrentMap<UUID, GridCommunicationClient> clients = U.field(spi, "clients"); assertEquals(1, clients.size()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java index 34fa610..c4a0916 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConfigSelfTest.java @@ -32,8 +32,6 @@ public class GridTcpCommunicationSpiConfigSelfTest extends GridSpiAbstractConfig checkNegativeSpiProperty(new TcpCommunicationSpi(), "localPort", 65636); checkNegativeSpiProperty(new TcpCommunicationSpi(), "localPortRange", -1); checkNegativeSpiProperty(new TcpCommunicationSpi(), "idleConnectionTimeout", 0); - checkNegativeSpiProperty(new TcpCommunicationSpi(), "connectionBufferSize", -1); - checkNegativeSpiProperty(new TcpCommunicationSpi(), "connectionBufferFlushFrequency", 0); checkNegativeSpiProperty(new TcpCommunicationSpi(), "socketReceiveBuffer", -1); checkNegativeSpiProperty(new TcpCommunicationSpi(), "socketSendBuffer", -1); checkNegativeSpiProperty(new TcpCommunicationSpi(), "messageQueueLimit", -1); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java index e7ae957..3916f02 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java @@ -501,7 +501,7 @@ public class GridTcpCommunicationSpiMultithreadedSelfTest extends GridSpiAbstrac } for (CommunicationSpi spi : spis.values()) { - final ConcurrentMap<UUID, GridTcpCommunicationClient> clients = U.field(spi, "clients"); + final ConcurrentMap<UUID, GridCommunicationClient> clients = U.field(spi, "clients"); assert GridTestUtils.waitForCondition(new PA() { @Override public boolean apply() { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java index 3c61f00..61bb944 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/AbstractDiscoverySelfTest.java @@ -22,9 +22,11 @@ import org.apache.ignite.cluster.*; import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.marshaller.*; import org.apache.ignite.spi.*; +import org.apache.ignite.testframework.*; import org.apache.ignite.testframework.config.*; import org.apache.ignite.testframework.junits.*; import org.apache.ignite.testframework.junits.spi.*; +import org.jetbrains.annotations.*; import javax.management.*; import java.io.*; @@ -132,7 +134,7 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri /** {@inheritDoc} */ @Override public void onDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode> topSnapshot, - Map<Long, Collection<ClusterNode>> topHist, Serializable data) { + Map<Long, Collection<ClusterNode>> topHist, @Nullable DiscoverySpiCustomMessage data) { if (type == EVT_NODE_METRICS_UPDATED) isMetricsUpdate = true; } @@ -205,7 +207,7 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri DiscoverySpiListener locHeartbeatLsnr = new DiscoverySpiListener() { @Override public void onDiscovery(int type, long topVer, ClusterNode node, Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>> topHist, - Serializable data) { + @Nullable DiscoverySpiCustomMessage data) { // If METRICS_UPDATED came from local node if (type == EVT_NODE_METRICS_UPDATED && node.id().equals(spi.getLocalNode().id())) @@ -266,9 +268,8 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri Collection<UUID> nodeIds = new HashSet<>(); - for (IgniteTestResources rsrc : spiRsrcs) { + for (IgniteTestResources rsrc : spiRsrcs) nodeIds.add(rsrc.getNodeId()); - } for (ClusterNode node : spi.getRemoteNodes()) { if (nodeIds.contains(node.id())) { @@ -369,7 +370,8 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri spi.setListener(new DiscoverySpiListener() { @SuppressWarnings({"NakedNotify"}) @Override public void onDiscovery(int type, long topVer, ClusterNode node, - Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>> topHist, Serializable data) { + Collection<ClusterNode> topSnapshot, Map<Long, Collection<ClusterNode>> topHist, + @Nullable DiscoverySpiCustomMessage data) { info("Discovery event [type=" + type + ", node=" + node + ']'); synchronized (mux) { @@ -388,6 +390,10 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri } }); + GridSpiTestContext ctx = initSpiContext(); + + GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "spiCtx", ctx); + spi.spiStart(getTestGridName() + i); spis.add(spi); @@ -395,7 +401,7 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri spiRsrcs.add(rsrcMgr); // Force to use test context instead of default dummy context. - spi.onContextInitialized(initSpiContext()); + spi.onContextInitialized(ctx); } } catch (Throwable e) { @@ -436,9 +442,8 @@ public abstract class AbstractDiscoverySelfTest<T extends IgniteSpi> extends Gri spi.spiStop(); } - for (IgniteTestResources rscrs : spiRsrcs) { + for (IgniteTestResources rscrs : spiRsrcs) rscrs.stopThreads(); - } // Clear. spis.clear(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/3d78aa15/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryMarshallerCheckSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryMarshallerCheckSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryMarshallerCheckSelfTest.java new file mode 100644 index 0000000..016854a --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryMarshallerCheckSelfTest.java @@ -0,0 +1,76 @@ +/* + * 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.spi.discovery.tcp; + +import org.apache.ignite.*; +import org.apache.ignite.configuration.*; +import org.apache.ignite.marshaller.jdk.*; +import org.apache.ignite.marshaller.optimized.*; +import org.apache.ignite.spi.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.*; +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*; +import org.apache.ignite.testframework.junits.common.*; + +/** + * Test for {@link org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi}. + */ +public class TcpClientDiscoveryMarshallerCheckSelfTest extends GridCommonAbstractTest { + /** */ + private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + if (gridName.endsWith("0")) + cfg.setMarshaller(new JdkMarshaller()); + else { + cfg.setClientMode(true); + + cfg.setMarshaller(new OptimizedMarshaller()); + } + + cfg.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(ipFinder)); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testMarshallerInConsistency() throws Exception { + startGrid(0); + + try { + startGrid(1); + + fail("Expected SPI exception was not thrown."); + } + catch (IgniteCheckedException e) { + Throwable ex = e.getCause().getCause(); + + assertTrue(ex instanceof IgniteSpiException); + assertTrue(ex.getMessage().contains("Local node's marshaller differs from remote node's marshaller")); + } + } +}
