ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [18/50] [abbrv] incubator-ignite git commit: Merge branch 'sprint-1' into ignite-128
Date Tue, 10 Feb 2015 15:00:29 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e8d42f5d/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
----------------------------------------------------------------------
diff --cc modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
index 5991ecc,0000000..f995356
mode 100644,000000..100644
--- a/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
+++ b/modules/clients/src/test/java/org/apache/ignite/internal/client/integration/ClientAbstractMultiNodeSelfTest.java
@@@ -1,857 -1,0 +1,860 @@@
 +/*
 + * 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.client.integration;
 +
 +import org.apache.ignite.*;
 +import org.apache.ignite.cache.*;
 +import org.apache.ignite.cluster.*;
 +import org.apache.ignite.compute.*;
 +import org.apache.ignite.configuration.*;
 +import org.apache.ignite.internal.*;
 +import org.apache.ignite.internal.client.*;
 +import org.apache.ignite.internal.client.balancer.*;
 +import org.apache.ignite.internal.client.ssl.*;
++import org.apache.ignite.internal.managers.communication.*;
 +import org.apache.ignite.internal.processors.cache.*;
 +import org.apache.ignite.internal.processors.cache.distributed.*;
++import org.apache.ignite.internal.processors.cache.transactions.*;
 +import org.apache.ignite.internal.processors.cache.version.*;
++import org.apache.ignite.internal.util.direct.*;
++import org.apache.ignite.internal.util.typedef.*;
 +import org.apache.ignite.lang.*;
 +import org.apache.ignite.resources.*;
 +import org.apache.ignite.spi.*;
 +import org.apache.ignite.spi.communication.tcp.*;
 +import org.apache.ignite.spi.discovery.tcp.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
 +import org.apache.ignite.internal.managers.communication.*;
 +import org.apache.ignite.internal.processors.cache.transactions.*;
 +import org.apache.ignite.internal.util.direct.*;
 +import org.apache.ignite.internal.util.typedef.*;
 +import org.apache.ignite.testframework.*;
 +import org.apache.ignite.testframework.junits.common.*;
 +import org.jetbrains.annotations.*;
 +
 +import java.util.*;
 +import java.util.concurrent.*;
 +import java.util.concurrent.atomic.*;
 +
 +import static java.util.concurrent.TimeUnit.*;
 +import static org.apache.ignite.cache.CacheAtomicityMode.*;
 +import static org.apache.ignite.cache.CacheDistributionMode.*;
 +import static org.apache.ignite.cache.CacheMode.*;
 +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
 +
 +/**
 + * Tests basic client behavior with multiple nodes.
 + */
 +@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
 +public abstract class ClientAbstractMultiNodeSelfTest extends GridCommonAbstractTest {
 +    /** */
 +    private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
 +
 +    /** Partitioned cache name. */
 +    private static final String PARTITIONED_CACHE_NAME = "partitioned";
 +
 +    /** Replicated cache name. */
 +    private static final String REPLICATED_CACHE_NAME = "replicated";
 +
 +    /** Replicated async cache name. */
 +    private static final String REPLICATED_ASYNC_CACHE_NAME = "replicated_async";
 +
 +    /** Nodes count. */
 +    public static final int NODES_CNT = 5;
 +
 +    /**
 +     * Topology update frequency.
 +     * Set it longer than router's, so we wouldn't receive failures
 +     * caused only by obsolete topology on router.
 +     */
 +    static final int TOP_REFRESH_FREQ = 2500;
 +
 +    /** Path to jetty config. */
 +    public static final String REST_JETTY_CFG = "modules/clients/src/test/resources/jetty/rest-jetty.xml";
 +
 +    /** Path to jetty config with SSl enabled. */
 +    public static final String REST_JETTY_SSL_CFG = "modules/clients/src/test/resources/jetty/rest-jetty-ssl.xml";
 +
 +    /** Host. */
 +    public static final String HOST = "127.0.0.1";
 +
 +    /** Base for tcp rest ports. */
 +    public static final int REST_TCP_PORT_BASE = 12345;
 +
 +    /** Base for http rest ports, defined in {@link #REST_JETTY_CFG}. */
 +    public static final int REST_HTTP_PORT_BASE = 11080;
 +
 +    /** Base for https rest ports, defined in {@link #REST_JETTY_SSL_CFG}. */
 +    public static final int REST_HTTPS_PORT_BASE = 11443;
 +
 +    /** */
 +    private static volatile boolean commSpiEnabled;
 +
 +    /** Flag to enable REST in node configuration. */
 +    private boolean restEnabled = true;
 +
 +    /** Client instance for each test. */
 +    private GridClient client;
 +
 +    /**
 +     * @return Client protocol that should be used.
 +     */
 +    protected abstract GridClientProtocol protocol();
 +
 +    /**
 +     * @return Server address to create first connection.
 +     */
 +    protected abstract String serverAddress();
 +
 +    /**
 +     * @return SSL context factory to use if SSL or {@code null} to disable SSL usage.
 +     */
 +    @Nullable protected GridSslContextFactory sslContextFactory() {
 +        return null;
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
 +        IgniteConfiguration c = super.getConfiguration(gridName);
 +
 +        c.setLocalHost(HOST);
 +
 +        assert c.getClientConnectionConfiguration() == null;
 +
 +        if (restEnabled) {
 +            ClientConnectionConfiguration clientCfg = new ClientConnectionConfiguration();
 +
 +            clientCfg.setRestTcpPort(REST_TCP_PORT_BASE);
 +
 +            GridSslContextFactory sslCtxFactory = sslContextFactory();
 +
 +            if (sslCtxFactory != null) {
 +                clientCfg.setRestTcpSslEnabled(true);
 +                clientCfg.setRestTcpSslContextFactory(sslCtxFactory);
 +            }
 +
 +            c.setClientConnectionConfiguration(clientCfg);
 +        }
 +
 +        TcpDiscoverySpi disco = new TcpDiscoverySpi();
 +
 +        disco.setIpFinder(IP_FINDER);
 +
 +        c.setDiscoverySpi(disco);
 +
 +        TestCommunicationSpi spi = new TestCommunicationSpi();
 +
 +        spi.setLocalPort(GridTestUtils.getNextCommPort(getClass()));
 +
 +        c.setCommunicationSpi(spi);
 +
 +        c.setCacheConfiguration(cacheConfiguration(null), cacheConfiguration(PARTITIONED_CACHE_NAME),
 +            cacheConfiguration(REPLICATED_CACHE_NAME), cacheConfiguration(REPLICATED_ASYNC_CACHE_NAME));
 +
 +        ThreadPoolExecutor exec = new ThreadPoolExecutor(
 +            40,
 +            40,
 +            0,
 +            MILLISECONDS,
 +            new LinkedBlockingQueue<Runnable>());
 +
 +        exec.prestartAllCoreThreads();
 +
 +        c.setExecutorService(exec);
 +
 +        c.setExecutorServiceShutdown(true);
 +
 +        ThreadPoolExecutor sysExec = new ThreadPoolExecutor(
 +            40,
 +            40,
 +            0,
 +            MILLISECONDS,
 +            new LinkedBlockingQueue<Runnable>());
 +
 +        sysExec.prestartAllCoreThreads();
 +
 +        c.setSystemExecutorService(sysExec);
 +
 +        c.setSystemExecutorServiceShutdown(true);
 +
 +        return c;
 +    }
 +
 +    /**
 +     * @param cacheName Cache name.
 +     * @return Cache configuration.
 +     * @throws Exception In case of error.
 +     */
 +    private CacheConfiguration cacheConfiguration(@Nullable String cacheName) throws Exception
{
 +        CacheConfiguration cfg = defaultCacheConfiguration();
 +
 +        cfg.setAtomicityMode(TRANSACTIONAL);
 +        cfg.setDistributionMode(NEAR_PARTITIONED);
 +
 +        if (cacheName == null)
 +            cfg.setCacheMode(LOCAL);
 +        else if (PARTITIONED_CACHE_NAME.equals(cacheName)) {
 +            cfg.setCacheMode(PARTITIONED);
 +
 +            cfg.setBackups(0);
 +        }
 +        else
 +            cfg.setCacheMode(REPLICATED);
 +
 +        cfg.setName(cacheName);
 +
 +        cfg.setWriteSynchronizationMode(REPLICATED_ASYNC_CACHE_NAME.equals(cacheName) ?
FULL_ASYNC : FULL_SYNC);
 +
 +        return cfg;
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void beforeTestsStarted() throws Exception {
 +        startGridsMultiThreaded(NODES_CNT);
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void afterTestsStopped() throws Exception {
 +        info("Stopping grids.");
 +
 +        stopAllGrids();
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void beforeTest() throws Exception {
 +        client = GridClientFactory.start(clientConfiguration());
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void afterTest() throws Exception {
 +        if (client != null) {
 +            GridClientFactory.stop(client.id(), false);
 +
 +            client = null;
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testSyncCommitRollbackFlags() throws Exception {
 +        commSpiEnabled = true;
 +
 +        try {
 +            GridClientData data = client.data(REPLICATED_ASYNC_CACHE_NAME);
 +
 +            info("Before put x1");
 +
 +            data.put("x1", "y1");
 +
 +            info("Before put x2");
 +
 +            data.flagsOn(GridClientCacheFlag.SYNC_COMMIT).put("x2", "y2");
 +
 +            info("Before put x3");
 +
 +            data.put("x3", "y3");
 +
 +            info("Before put x4");
 +
 +            data.flagsOn(GridClientCacheFlag.SYNC_COMMIT).put("x4", "y4");
 +        }
 +        finally {
 +            commSpiEnabled = false;
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testEmptyProjections() throws Exception {
 +        final GridClientCompute dflt = client.compute();
 +
 +        Collection<? extends GridClientNode> nodes = dflt.nodes();
 +
 +        assertEquals(NODES_CNT, nodes.size());
 +
 +        Iterator<? extends GridClientNode> iter = nodes.iterator();
 +
 +        final GridClientCompute singleNodePrj = dflt.projection(Collections.singletonList(iter.next()));
 +
 +        final GridClientNode second = iter.next();
 +
 +        final GridClientPredicate<GridClientNode> noneFilter = new GridClientPredicate<GridClientNode>()
{
 +            @Override public boolean apply(GridClientNode node) {
 +                return false;
 +            }
 +        };
 +
 +        final GridClientPredicate<GridClientNode> targetFilter = new GridClientPredicate<GridClientNode>()
{
 +            @Override public boolean apply(GridClientNode node) {
 +                return node.nodeId().equals(second.nodeId());
 +            }
 +        };
 +
 +        GridTestUtils.assertThrows(log(), new Callable<Object>() {
 +            @Override public Object call() throws Exception {
 +                return dflt.projection(noneFilter).log(-1, -1);
 +            }
 +        }, GridServerUnreachableException.class, null);
 +
 +        GridTestUtils.assertThrows(log(), new Callable<Object>() {
 +            @Override public Object call() throws Exception {
 +                return singleNodePrj.projection(second);
 +            }
 +        }, GridClientException.class, null);
 +
 +        GridTestUtils.assertThrows(log(), new Callable<Object>() {
 +            @Override
 +            public Object call() throws Exception {
 +                return singleNodePrj.projection(targetFilter);
 +            }
 +        }, GridClientException.class, null);
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testProjectionRun() throws Exception {
 +        GridClientCompute dflt = client.compute();
 +
 +        Collection<? extends GridClientNode> nodes = dflt.nodes();
 +
 +        assertEquals(NODES_CNT, nodes.size());
 +
 +        for (int i = 0; i < NODES_CNT; i++) {
 +            Ignite g = grid(i);
 +
 +            assert g != null;
 +
 +            GridClientNode clientNode = dflt.node(g.cluster().localNode().id());
 +
 +            assertNotNull("Client node for " + g.cluster().localNode().id() + " was not
found", clientNode);
 +
 +            GridClientCompute prj = dflt.projection(clientNode);
 +
 +            String res = prj.execute(TestTask.class.getName(), null);
 +
 +            assertNotNull(res);
 +
 +            assertEquals(g.cluster().localNode().id().toString(), res);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testAffinityExecute() throws Exception {
 +        GridClientCompute dflt = client.compute();
 +
 +        GridClientData data = client.data(PARTITIONED_CACHE_NAME);
 +
 +        Collection<? extends GridClientNode> nodes = dflt.nodes();
 +
 +        assertEquals(NODES_CNT, nodes.size());
 +
 +        for (int i = 0; i < NODES_CNT; i++) {
 +            Ignite g = grid(i);
 +
 +            assert g != null;
 +
 +            int affinityKey = -1;
 +
 +            for (int key = 0; key < 10000; key++) {
 +                if (g.cluster().localNode().id().equals(data.affinity(key))) {
 +                    affinityKey = key;
 +
 +                    break;
 +                }
 +            }
 +
 +            if (affinityKey == -1)
 +                throw new Exception("Unable to found key for which node is primary: " +
g.cluster().localNode().id());
 +
 +            GridClientNode clientNode = dflt.node(g.cluster().localNode().id());
 +
 +            assertNotNull("Client node for " + g.cluster().localNode().id() + " was not
found", clientNode);
 +
 +            String res = dflt.affinityExecute(TestTask.class.getName(), PARTITIONED_CACHE_NAME,
affinityKey, null);
 +
 +            assertNotNull(res);
 +
 +            assertEquals(g.cluster().localNode().id().toString(), res);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testInvalidateFlag() throws Exception {
 +        IgniteEx g0 = grid(0);
 +
 +        GridCache<String, String> cache = g0.cache(PARTITIONED_CACHE_NAME);
 +
 +        String key = null;
 +
 +        for (int i = 0; i < 10_000; i++) {
 +            if (!cache.affinity().isPrimaryOrBackup(g0.localNode(), String.valueOf(i)))
{
 +                key = String.valueOf(i);
 +
 +                break;
 +            }
 +        }
 +
 +        assertNotNull(key);
 +
 +        cache.put(key, key); // Create entry in near cache, it is invalidated if INVALIDATE
flag is set.
 +
 +        assertNotNull(cache.peek(key));
 +
 +        GridClientData d = client.data(PARTITIONED_CACHE_NAME);
 +
 +        d.flagsOn(GridClientCacheFlag.INVALIDATE).put(key, "zzz");
 +
 +        for (Ignite g : G.allGrids()) {
 +            cache = g.cache(PARTITIONED_CACHE_NAME);
 +
 +            if (cache.affinity().isPrimaryOrBackup(g.cluster().localNode(), key))
 +                assertEquals("zzz", cache.peek(key));
 +            else
 +                assertNull(cache.peek(key));
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testClientAffinity() throws Exception {
 +        GridClientData partitioned = client.data(PARTITIONED_CACHE_NAME);
 +
 +        Collection<Object> keys = new ArrayList<>();
 +
 +        keys.addAll(Arrays.asList(
 +            Boolean.TRUE,
 +            Boolean.FALSE,
 +            1,
 +            Integer.MAX_VALUE
 +        ));
 +
 +        Random rnd = new Random();
 +        StringBuilder sb = new StringBuilder();
 +
 +        // Generate some random strings.
 +        for (int i = 0; i < 100; i++) {
 +            sb.setLength(0);
 +
 +            for (int j = 0; j < 255; j++)
 +                // Only printable ASCII symbols for test.
 +                sb.append((char)(rnd.nextInt(0x7f - 0x20) + 0x20));
 +
 +            keys.add(sb.toString());
 +        }
 +
 +        // Generate some more keys to achieve better coverage.
 +        for (int i = 0; i < 100; i++)
 +            keys.add(UUID.randomUUID());
 +
 +        for (Object key : keys) {
 +            UUID nodeId = grid(0).mapKeyToNode(PARTITIONED_CACHE_NAME, key).id();
 +
 +            UUID clientNodeId = partitioned.affinity(key);
 +
 +            assertEquals("Invalid affinity mapping for REST response for key: " + key, nodeId,
clientNodeId);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testTopologyListener() throws Exception {
 +        final Collection<UUID> added = new ArrayList<>(1);
 +        final Collection<UUID> rmvd = new ArrayList<>(1);
 +
 +        final CountDownLatch addedLatch = new CountDownLatch(1);
 +        final CountDownLatch rmvLatch = new CountDownLatch(1);
 +
 +        assertEquals(NODES_CNT, client.compute().refreshTopology(false, false).size());
 +
 +        GridClientTopologyListener lsnr = new GridClientTopologyListener() {
 +            @Override public void onNodeAdded(GridClientNode node) {
 +                added.add(node.nodeId());
 +
 +                addedLatch.countDown();
 +            }
 +
 +            @Override public void onNodeRemoved(GridClientNode node) {
 +                rmvd.add(node.nodeId());
 +
 +                rmvLatch.countDown();
 +            }
 +        };
 +
 +        client.addTopologyListener(lsnr);
 +
 +        try {
 +            Ignite g = startGrid(NODES_CNT + 1);
 +
 +            UUID id = g.cluster().localNode().id();
 +
 +            assertTrue(addedLatch.await(2 * TOP_REFRESH_FREQ, MILLISECONDS));
 +
 +            assertEquals(1, added.size());
 +            assertEquals(id, F.first(added));
 +
 +            stopGrid(NODES_CNT + 1);
 +
 +            assertTrue(rmvLatch.await(2 * TOP_REFRESH_FREQ, MILLISECONDS));
 +
 +            assertEquals(1, rmvd.size());
 +            assertEquals(id, F.first(rmvd));
 +        }
 +        finally {
 +            client.removeTopologyListener(lsnr);
 +
 +            stopGrid(NODES_CNT + 1);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testDisabledRest() throws Exception {
 +        restEnabled = false;
 +
 +        final Ignite g = startGrid("disabled-rest");
 +
 +        try {
 +            Thread.sleep(2 * TOP_REFRESH_FREQ);
 +
 +            // As long as we have round robin load balancer this will cause every node to
be queried.
 +            for (int i = 0; i < NODES_CNT + 1; i++)
 +                assertEquals(NODES_CNT + 1, client.compute().refreshTopology(false, false).size());
 +
 +            final GridClientData data = client.data(PARTITIONED_CACHE_NAME);
 +
 +            // Check rest-disabled node is unavailable.
 +            try {
 +                String affKey;
 +
 +                do {
 +                    affKey = UUID.randomUUID().toString();
 +                } while (!data.affinity(affKey).equals(g.cluster().localNode().id()));
 +
 +                data.put(affKey, "asdf");
 +
 +                assertEquals("asdf", cache(0, PARTITIONED_CACHE_NAME).get(affKey));
 +            }
 +            catch (GridServerUnreachableException e) {
 +                // Thrown for direct client-node connections.
 +                assertTrue("Unexpected exception message: " + e.getMessage(),
 +                    e.getMessage().startsWith("No available endpoints to connect (is rest
enabled for this node?)"));
 +            }
 +            catch (GridClientException e) {
 +                // Thrown for routed client-router-node connections.
 +                String msg = e.getMessage();
 +
 +                assertTrue("Unexpected exception message: " + msg, protocol() == GridClientProtocol.TCP
?
 +                    msg.contains("No available endpoints to connect (is rest enabled for
this node?)") : // TCP router.
 +                    msg.startsWith("No available nodes on the router for destination node
ID"));         // HTTP router.
 +            }
 +
 +            // Check rest-enabled nodes are available.
 +            String affKey;
 +
 +            do {
 +                affKey = UUID.randomUUID().toString();
 +            } while (data.affinity(affKey).equals(g.cluster().localNode().id()));
 +
 +            data.put(affKey, "fdsa");
 +
 +            assertEquals("fdsa", cache(0, PARTITIONED_CACHE_NAME).get(affKey));
 +        }
 +        finally {
 +            restEnabled = true;
 +
 +            G.stop(g.name(), true);
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testAffinityPut() throws Exception {
 +        Thread.sleep(2 * TOP_REFRESH_FREQ);
 +
 +        assertEquals(NODES_CNT, client.compute().refreshTopology(false, false).size());
 +
 +        Map<UUID, Ignite> gridsByLocNode = new HashMap<>(NODES_CNT);
 +
 +        GridClientData partitioned = client.data(PARTITIONED_CACHE_NAME);
 +
 +        GridClientCompute compute = client.compute();
 +
 +        for (int i = 0; i < NODES_CNT; i++)
 +            gridsByLocNode.put(grid(i).localNode().id(), grid(i));
 +
 +        for (int i = 0; i < 100; i++) {
 +            String key = "key" + i;
 +
 +            UUID primaryNodeId = grid(0).mapKeyToNode(PARTITIONED_CACHE_NAME, key).id();
 +
 +            assertEquals("Affinity mismatch for key: " + key, primaryNodeId, partitioned.affinity(key));
 +
 +            assertEquals(primaryNodeId, partitioned.affinity(key));
 +
 +            // Must go to primary node only. Since backup count is 0, value must present
on
 +            // primary node only.
 +            partitioned.put(key, "val" + key);
 +
 +            for (Map.Entry<UUID, Ignite> entry : gridsByLocNode.entrySet()) {
 +                Object val = entry.getValue().cache(PARTITIONED_CACHE_NAME).peek(key);
 +
 +                if (primaryNodeId.equals(entry.getKey()))
 +                    assertEquals("val" + key, val);
 +                else
 +                    assertNull(val);
 +            }
 +        }
 +
 +        // Now check that we will see value in near cache in pinned mode.
 +        for (int i = 100; i < 200; i++) {
 +            String pinnedKey = "key" + i;
 +
 +            UUID primaryNodeId = grid(0).mapKeyToNode(PARTITIONED_CACHE_NAME, pinnedKey).id();
 +
 +            UUID pinnedNodeId = F.first(F.view(gridsByLocNode.keySet(), F.notEqualTo(primaryNodeId)));
 +
 +            GridClientNode node = compute.node(pinnedNodeId);
 +
 +            partitioned.pinNodes(node).put(pinnedKey, "val" + pinnedKey);
 +
 +            for (Map.Entry<UUID, Ignite> entry : gridsByLocNode.entrySet()) {
 +                Object val = entry.getValue().cache(PARTITIONED_CACHE_NAME).peek(pinnedKey);
 +
 +                if (primaryNodeId.equals(entry.getKey()) || pinnedNodeId.equals(entry.getKey()))
 +                    assertEquals("val" + pinnedKey, val);
 +                else
 +                    assertNull(val);
 +            }
 +        }
 +    }
 +
 +    /**
 +     * @return Client configuration for the test.
 +     */
 +    protected GridClientConfiguration clientConfiguration() throws GridClientException {
 +        GridClientConfiguration cfg = new GridClientConfiguration();
 +
 +        cfg.setBalancer(getBalancer());
 +
 +        cfg.setTopologyRefreshFrequency(TOP_REFRESH_FREQ);
 +
 +        cfg.setProtocol(protocol());
 +        cfg.setServers(Arrays.asList(serverAddress()));
 +        cfg.setSslContextFactory(sslContextFactory());
 +
 +        GridClientDataConfiguration loc = new GridClientDataConfiguration();
 +
 +        GridClientDataConfiguration partitioned = new GridClientDataConfiguration();
 +
 +        partitioned.setName(PARTITIONED_CACHE_NAME);
 +        partitioned.setAffinity(new GridClientPartitionAffinity());
 +
 +        GridClientDataConfiguration replicated = new GridClientDataConfiguration();
 +        replicated.setName(REPLICATED_CACHE_NAME);
 +
 +        GridClientDataConfiguration replicatedAsync = new GridClientDataConfiguration();
 +        replicatedAsync.setName(REPLICATED_ASYNC_CACHE_NAME);
 +
 +        cfg.setDataConfigurations(Arrays.asList(loc, partitioned, replicated, replicatedAsync));
 +
 +        return cfg;
 +    }
 +
 +    /**
 +     * Gets client load balancer.
 +     *
 +     * @return Load balancer.
 +     */
 +    protected GridClientLoadBalancer getBalancer() {
 +        return new GridClientRoundRobinBalancer();
 +    }
 +
 +    /**
 +     * Test task. Returns a tuple in which first component is id of node that has split
the task,
 +     * and second component is count of nodes that executed jobs.
 +     */
 +    private static class TestTask extends ComputeTaskSplitAdapter<Object, String>
{
 +        /** */
 +        @IgniteInstanceResource
 +        private Ignite ignite;
 +
 +        /** Count of tasks this job was split to. */
 +        private int gridSize;
 +
 +        /** {@inheritDoc} */
-         @Override protected Collection<? extends ComputeJob> split(int gridSize, Object
arg)
-             throws IgniteCheckedException {
++        @Override protected Collection<? extends ComputeJob> split(int gridSize, Object
arg) {
 +            Collection<ComputeJobAdapter> jobs = new ArrayList<>(gridSize);
 +
 +            this.gridSize = gridSize;
 +
 +            final String locNodeId = ignite.cluster().localNode().id().toString();
 +
 +            for (int i = 0; i < gridSize; i++) {
 +                jobs.add(new ComputeJobAdapter() {
 +                    @SuppressWarnings("OverlyStrongTypeCast")
 +                    @Override public Object execute() {
 +                        try {
 +                            Thread.sleep(1000);
 +                        }
 +                        catch (InterruptedException ignored) {
 +                            Thread.currentThread().interrupt();
 +                        }
 +
 +                        return new IgniteBiTuple<>(locNodeId, 1);
 +                    }
 +                });
 +            }
 +
 +            return jobs;
 +        }
 +
 +        /** {@inheritDoc} */
-         @Override public String reduce(List<ComputeJobResult> results) throws IgniteCheckedException
{
++        @Override public String reduce(List<ComputeJobResult> results) {
 +            int sum = 0;
 +
 +            String locNodeId = null;
 +
 +            for (ComputeJobResult res : results) {
 +                IgniteBiTuple<String, Integer> part = res.getData();
 +
 +                if (locNodeId == null)
 +                    locNodeId = part.get1();
 +
 +                Integer i = part.get2();
 +
 +                if (i != null)
 +                    sum += i;
 +            }
 +
 +            assert gridSize == sum;
 +
 +            return locNodeId;
 +        }
 +    }
 +
 +    /**
 +     * Communication SPI which checks cache flags.
 +     */
 +    @SuppressWarnings("unchecked")
 +    private static class TestCommunicationSpi extends TcpCommunicationSpi {
 +        /** {@inheritDoc} */
 +        @Override public void sendMessage(ClusterNode node, GridTcpCommunicationMessageAdapter
msg)
 +            throws IgniteSpiException {
 +            checkSyncFlags((GridIoMessage)msg);
 +
 +            super.sendMessage(node, msg);
 +        }
 +
 +        /**
 +         * Check if flags in correct state.
 +         *
 +         * @param msg Message.
 +         */
 +        private void checkSyncFlags(GridIoMessage msg) {
 +            if (!commSpiEnabled)
 +                return;
 +
 +            Object o = msg.message();
 +
 +            if (!(o instanceof GridDistributedLockRequest))
 +                return;
 +
 +            IgniteKernal g = (IgniteKernal)G.ignite(ignite.configuration().getNodeId());
 +
 +            GridCacheContext<Object, Object> cacheCtx = g.internalCache(REPLICATED_ASYNC_CACHE_NAME).context();
 +
 +            IgniteTxManager<Object, Object> tm = cacheCtx.tm();
 +
 +            GridCacheVersion v = ((GridCacheVersionable)o).version();
 +
-             IgniteTxEx t = tm.tx(v);
++            IgniteInternalTx t = tm.tx(v);
 +
 +            if (t.hasWriteKey(cacheCtx.txKey("x1")))
 +                assertFalse("Invalid tx flags: " + t, t.syncCommit());
 +            else if (t.hasWriteKey(cacheCtx.txKey("x2")))
 +                assertTrue("Invalid tx flags: " + t, t.syncCommit());
 +            else if (t.hasWriteKey(cacheCtx.txKey("x3")))
 +                assertFalse("Invalid tx flags: " + t, t.syncCommit());
 +            else if (t.hasWriteKey(cacheCtx.txKey("x4")))
 +                assertTrue("Invalid tx flags: " + t, t.syncCommit());
 +        }
 +    }
 +
 +    /**
 +     * @throws Exception If failed.
 +     */
 +    public void testMultithreadedCommand() throws Exception {
 +        final GridClientData data = client.data(PARTITIONED_CACHE_NAME);
 +        final GridClientCompute compute = client.compute();
 +        final AtomicInteger cnt = new AtomicInteger(0);
 +
 +        multithreaded(new Callable<Object>() {
 +            @Override public Object call() throws Exception {
 +                for (int i = 0; i < 20; i++) {
 +                    String key = UUID.randomUUID().toString();
 +                    String val = UUID.randomUUID().toString();
 +
 +                    switch (cnt.incrementAndGet() % 4) {
 +                        case 0: {
 +                            assertTrue(data.put(key, val));
 +                            assertEquals(val, data.get(key));
 +                            assertTrue(data.remove(key));
 +
 +                            break;
 +                        }
 +
 +                        case 1: {
 +                            assertNotNull(data.metrics());
 +
 +                            break;
 +                        }
 +
 +                        case 2: {
 +                            String nodeId = compute.execute(TestTask.class.getName(), null);
 +
 +                            assertNotNull(nodeId);
 +                            assertNotNull(compute.refreshNode(UUID.fromString(nodeId), true,
true));
 +
 +                            break;
 +                        }
 +
 +                        case 3: {
 +                            assertEquals(NODES_CNT, compute.refreshTopology(true, true).size());
 +
 +                            break;
 +                        }
 +                    }
 +                }
 +
 +                return null;
 +            }
 +        }, 50, "multithreaded-client-access");
 +    }
 +}


Mime
View raw message