ThinClientPartitionAwarenessStableTopologyTest.java

/*
 * 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.thin;

import java.util.function.Function;

import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.client.ClientAtomicConfiguration;
import org.apache.ignite.client.ClientAtomicLong;
import org.apache.ignite.client.ClientCache;
import org.apache.ignite.configuration.AtomicConfiguration;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.datastructures.GridCacheAtomicLongEx;
import org.junit.Test;

/**
 * Test partition awareness of thin client on stable topology.
 */
public class ThinClientPartitionAwarenessStableTopologyTest extends ThinClientAbstractPartitionAwarenessTest {
    /** Grids count. */
    private static final int GRIDS_CNT = 3;

    /** {@inheritDoc} */
    @Override protected void beforeTestsStarted() throws Exception {
        super.beforeTestsStarted();

        startGrids(GRIDS_CNT);

        awaitPartitionMapExchange();
    }

    /** {@inheritDoc} */
    @Override protected void beforeTest() throws Exception {
        super.beforeTest();

        // Add one extra node address to the list, skip 0 node.
        initClient(getClientConfiguration(1, 2, 3), 1, 2);
    }

    /**
     * Test that partition awareness is not applicable for replicated cache.
     */
    @Test
    public void testReplicatedCache() {
        testNotApplicableCache(REPL_CACHE_NAME);
    }

    /**
     * Test that partition awareness is not applicable for partitioned cache with custom affinity function.
     */
    @Test
    public void testPartitionedCustomAffinityCache() {
        testNotApplicableCache(PART_CUSTOM_AFFINITY_CACHE_NAME);
    }

    /**
     * Test partition awareness for all applicable operation types for partitioned cache with primitive key.
     */
    @Test
    public void testPartitionedCachePrimitiveKey() throws Exception {
        testApplicableCache(PART_CACHE_NAME, i -> i);
    }

    /**
     * Test partition awareness for all applicable operation types for partitioned cache with complex key.
     */
    @Test
    public void testPartitionedCacheComplexKey() throws Exception {
        testApplicableCache(PART_CACHE_NAME, i -> new TestComplexKey(i, i));
    }

    /**
     * Test partition awareness for all applicable operation types for partitioned cache with annotated affinity
     * mapped key.
     */
    @Test
    public void testPartitionedCacheAnnotatedAffinityKey() throws Exception {
        testApplicableCache(PART_CACHE_NAME, i -> new TestAnnotatedAffinityKey(i, i));
    }

    /**
     * Test partition awareness for all applicable operation types for partitioned cache with not annotated affinity
     * mapped key.
     */
    @Test
    public void testPartitionedCacheNotAnnotatedAffinityKey() throws Exception {
        testApplicableCache(PART_CACHE_NAME, i -> new TestNotAnnotatedAffinityKey(new TestComplexKey(i, i), i));
    }

    /**
     * Test request to partition mapped to unknown for client node.
     */
    @Test
    public void testPartitionedCacheUnknownNode() throws IgniteCheckedException {
        ClientCache<Object, Object> clientCache = client.cache(PART_CACHE_NAME);

        // We don't included grid(0) address to list of addresses known for the client, so client don't have connection
        // with grid(0)
        Integer keyForUnknownNode = primaryKey(grid(0).cache(PART_CACHE_NAME));

        assertNotNull("Not found key for node " + grid(0).localNode().id(), keyForUnknownNode);

        clientCache.put(keyForUnknownNode, 0);

        assertOpOnChannel(dfltCh, ClientOperation.CACHE_PARTITIONS);
        assertOpOnChannel(dfltCh, ClientOperation.CACHE_PUT);
    }

    /**
     * Test affinity awareness for all applicable operation types for partitioned cache with 0 backups.
     */
    @Test
    public void testPartitionedCache0Backups() throws Exception {
        testApplicableCache(PART_CACHE_0_BACKUPS_NAME, i -> i);
    }

    /**
     * Test affinity awareness for all applicable operation types for partitioned cache with 1 backups.
     */
    @Test
    public void testPartitionedCache1Backups() throws Exception {
        testApplicableCache(PART_CACHE_1_BACKUPS_NAME, i -> i);
    }

    /**
     * Test affinity awareness for all applicable operation types for partitioned cache with 3 backups.
     */
    @Test
    public void testPartitionedCache3Backups() throws Exception {
        testApplicableCache(PART_CACHE_3_BACKUPS_NAME, i -> i);
    }

    /**
     * Test scan query.
     */
    @Test
    public void testScanQuery() throws IgniteCheckedException {
        ClientCache<Object, Object> clientCache = client.cache(PART_CACHE_NAME);

        // Make any operation to request partitions.
        clientCache.get(0);

        opsQueue.clear(); // Clear partitions request and get operation.

        for (int i = 0; i < GRIDS_CNT; i++) {
            int part = grid(i).affinity(PART_CACHE_NAME).primaryPartitions(grid(i).localNode())[0];

            // Client doesn't have connection with grid(0).
            TestTcpClientChannel ch = i == 0 ? dfltCh : nodeChannel(grid(i).localNode().id());

            // Test scan query with specified partition.
            clientCache.query(new ScanQuery<>().setPartition(part)).getAll();

            assertOpOnChannel(ch, ClientOperation.QUERY_SCAN);

            // Test scan query without specified partition.
            clientCache.query(new ScanQuery<>()).getAll();

            assertOpOnChannel(dfltCh, ClientOperation.QUERY_SCAN);
        }
    }

    /**
     * Test atomic long.
     */
    @Test
    public void testAtomicLong() {
        testAtomicLong("default-grp-partitioned", null, CacheMode.PARTITIONED);
        testAtomicLong("default-grp-replicated", null, CacheMode.REPLICATED);
        testAtomicLong("custom-grp-partitioned", "testAtomicLong", CacheMode.PARTITIONED);
        testAtomicLong("custom-grp-replicated", "testAtomicLong", CacheMode.REPLICATED);
    }

    /**
     * Test atomic long.
     */
    private void testAtomicLong(String name, String grpName, CacheMode cacheMode) {
        ClientAtomicConfiguration cfg = new ClientAtomicConfiguration()
                .setGroupName(grpName)
                .setCacheMode(cacheMode);

        ClientAtomicLong clientAtomicLong = client.atomicLong(name, cfg, 1, true);
        GridCacheAtomicLongEx serverAtomicLong = (GridCacheAtomicLongEx)grid(0).atomicLong(
                name, new AtomicConfiguration().setGroupName(grpName), 0, false);

        String cacheName = "ignite-sys-atomic-cache@" + (grpName == null ? "default-ds-group" : grpName);
        IgniteInternalCache<Object, Object> cache = grid(0).context().cache().cache(cacheName);

        // Warm up.
        clientAtomicLong.get();
        opsQueue.clear();

        // Test.
        clientAtomicLong.get();
        TestTcpClientChannel opCh = affinityChannel(serverAtomicLong.key(), cache);

        assertOpOnChannel(opCh, ClientOperation.ATOMIC_LONG_VALUE_GET);
    }

    /**
     * @param cacheName Cache name.
     */
    private void testNotApplicableCache(String cacheName) {
        ClientCache<Object, Object> cache = client.cache(cacheName);

        // After first response we should send partitions request on default channel together with next request.
        cache.put(0, 0);

        assertOpOnChannel(dfltCh, ClientOperation.CACHE_PARTITIONS);
        assertOpOnChannel(dfltCh, ClientOperation.CACHE_PUT);

        for (int i = 1; i < KEY_CNT; i++) {
            cache.put(i, i);

            assertOpOnChannel(dfltCh, ClientOperation.CACHE_PUT);

            cache.get(i);

            assertOpOnChannel(dfltCh, ClientOperation.CACHE_GET);
        }
    }

    /**
     * @param cacheName Cache name.
     * @param keyFactory Key factory function.
     */
    private void testApplicableCache(String cacheName, Function<Integer, Object> keyFactory) throws Exception {
        ClientCache<Object, Object> clientCache = client.cache(cacheName);
        IgniteInternalCache<Object, Object> igniteCache = grid(0).context().cache().cache(cacheName);

        clientCache.put(keyFactory.apply(0), 0);

        TestTcpClientChannel opCh = affinityChannel(keyFactory.apply(0), igniteCache);

        // Default channel is the first who detects topology change, so next partition request will go through
        // the default channel.
        assertOpOnChannel(dfltCh, ClientOperation.CACHE_PARTITIONS);
        assertOpOnChannel(opCh, ClientOperation.CACHE_PUT);

        for (int i = 1; i < KEY_CNT; i++) {
            Object key = keyFactory.apply(i);

            opCh = affinityChannel(key, igniteCache);

            clientCache.put(key, key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_PUT);

            clientCache.putAsync(key, key).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_PUT);

            clientCache.get(key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET);

            clientCache.getAsync(key).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET);

            clientCache.containsKey(key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_CONTAINS_KEY);

            clientCache.containsKeyAsync(key).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_CONTAINS_KEY);

            clientCache.replace(key, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_REPLACE);

            clientCache.replaceAsync(key, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_REPLACE);

            clientCache.replace(key, i, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_REPLACE_IF_EQUALS);

            clientCache.replaceAsync(key, i, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_REPLACE_IF_EQUALS);

            clientCache.remove(key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_REMOVE_KEY);

            clientCache.removeAsync(key).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_REMOVE_KEY);

            clientCache.remove(key, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_REMOVE_IF_EQUALS);

            clientCache.removeAsync(key, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_REMOVE_IF_EQUALS);

            clientCache.getAndPut(key, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_PUT);

            clientCache.getAndPutAsync(key, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_PUT);

            clientCache.getAndRemove(key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_REMOVE);

            clientCache.getAndRemoveAsync(key).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_REMOVE);

            clientCache.getAndReplace(key, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_REPLACE);

            clientCache.getAndReplaceAsync(key, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_REPLACE);

            clientCache.putIfAbsent(key, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_PUT_IF_ABSENT);

            clientCache.putIfAbsentAsync(key, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_PUT_IF_ABSENT);

            clientCache.getAndPutIfAbsent(key, i);
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_PUT_IF_ABSENT);

            clientCache.getAndPutIfAbsentAsync(key, i).get();
            assertOpOnChannel(opCh, ClientOperation.CACHE_GET_AND_PUT_IF_ABSENT);

            clientCache.clear(key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_CLEAR_KEY);

            clientCache.clearAsync(key);
            assertOpOnChannel(opCh, ClientOperation.CACHE_CLEAR_KEY);
        }
    }
}