<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.cache.distributed.near;

import org.apache.ignite.*;
import org.apache.ignite.cache.affinity.*;
import org.apache.ignite.cache.affinity.rendezvous.*;
import org.apache.ignite.cache.query.*;
import org.apache.ignite.cache.query.annotations.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.typedef.*;
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 javax.cache.*;
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import static org.apache.ignite.cache.CacheAtomicityMode.*;
import static org.apache.ignite.cache.CacheMode.*;
import static org.apache.ignite.cache.CacheRebalanceMode.*;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;

/**
 * Test for distributed queries with replicated client cache and node restarts.
 */
public class IgniteCacheClientQueryReplicatedNodeRestartSelfTest extends GridCommonAbstractTest {
    /** */
    private static final String QRY = "select co.id, count(*) cnt\n" +
        "from \"pe\".Person pe, \"pr\".Product pr, \"co\".Company co, \"pu\".Purchase pu\n" +
        "where pe.id = pu.personId and pu.productId = pr.id and pr.companyId = co.id \n" +
        "group by co.id order by cnt desc, co.id";

    /** */
    private static final P1<ClusterNode> DATA_NODES_FILTER = new P1<ClusterNode>() {
            @Override public boolean apply(ClusterNode clusterNode) {
                String gridName = clusterNode.attribute(IgniteNodeAttributes.ATTR_GRID_NAME);

                return !gridName.endsWith(String.valueOf(GRID_CNT - 1)); // The last one is client only.
            }
        };

    /** */
    private static final List<List<?>> FAKE = new LinkedList<>();

    /** */
    private static final int GRID_CNT = 5;

    /** */
    private static final int PERS_CNT = 600;

    /** */
    private static final int PURCHASE_CNT = 6000;

    /** */
    private static final int COMPANY_CNT = 25;

    /** */
    private static final int PRODUCT_CNT = 100;

    /** */
    private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);

    /** {@inheritDoc} */
    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
        X.println("grid name: " + gridName);

        IgniteConfiguration c = super.getConfiguration(gridName);

        TcpDiscoverySpi disco = new TcpDiscoverySpi();

        disco.setIpFinder(ipFinder);

        c.setDiscoverySpi(disco);

        int i = 0;

        CacheConfiguration<?, ?>[] ccs = new CacheConfiguration[4];

        for (String name : F.asList("co", "pr", "pe", "pu")) {
            CacheConfiguration<?, ?> cc = defaultCacheConfiguration();

            cc.setNodeFilter(DATA_NODES_FILTER);
            cc.setName(name);
            cc.setCacheMode(REPLICATED);
            cc.setWriteSynchronizationMode(FULL_SYNC);
            cc.setAtomicityMode(TRANSACTIONAL);
            cc.setRebalanceMode(SYNC);
            cc.setAffinity(new RendezvousAffinityFunction(false, 50));

            switch (name) {
                case "co":
                    cc.setIndexedTypes(
                        Integer.class, Company.class
                    );

                    break;

                case "pr":
                    cc.setIndexedTypes(
                        Integer.class, Product.class
                    );

                    break;

                case "pe":
                    cc.setIndexedTypes(
                        Integer.class, Person.class
                    );

                    break;

                case "pu":
                    cc.setIndexedTypes(
                        AffinityKey.class, Purchase.class
                    );

                    break;
            }

            ccs[i++] = cc;
        }

        c.setCacheConfiguration(ccs);

        return c;
    }

    /**
     *
     */
    private void fillCaches() {
        IgniteCache<Integer, Company> co = grid(0).cache("co");

        for (int i = 0; i < COMPANY_CNT; i++)
            co.put(i, new Company(i));

        IgniteCache<Integer, Product> pr = grid(0).cache("pr");

        Random rnd = new GridRandom();

        for (int i = 0; i < PRODUCT_CNT; i++)
            pr.put(i, new Product(i, rnd.nextInt(COMPANY_CNT)));

        IgniteCache<Integer, Person> pe = grid(0).cache("pe");

        for (int i = 0; i < PERS_CNT; i++)
            pe.put(i, new Person(i));

        IgniteCache<AffinityKey<Integer>, Purchase> pu = grid(0).cache("pu");

        for (int i = 0; i < PURCHASE_CNT; i++) {
            int persId = rnd.nextInt(PERS_CNT);
            int prodId = rnd.nextInt(PRODUCT_CNT);

            pu.put(new AffinityKey<>(i, persId), new Purchase(persId, prodId));
        }
    }

    /**
     * @param c Cache.
     * @param client If it must be a client cache.
     */
    private void assertClient(IgniteCache<?,?> c, boolean client) {
        assertTrue(((IgniteCacheProxy)c).context().affinityNode() == !client);
    }

    /**
     * @throws Exception If failed.
     */
    public void testRestarts() throws Exception {
        int duration = 90 * 1000;
        int qryThreadNum = 5;
        int restartThreadsNum = 3; // 3 of 4 data nodes<commit_msg>ignite-484-1 - relax test: 2 restarting nodes instead of 3<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.cache.distributed.near;

import org.apache.ignite.*;
import org.apache.ignite.cache.affinity.*;
import org.apache.ignite.cache.affinity.rendezvous.*;
import org.apache.ignite.cache.query.*;
import org.apache.ignite.cache.query.annotations.*;
import org.apache.ignite.cluster.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.processors.cache.*;
import org.apache.ignite.internal.util.*;
import org.apache.ignite.internal.util.typedef.*;
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 javax.cache.*;
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import static org.apache.ignite.cache.CacheAtomicityMode.*;
import static org.apache.ignite.cache.CacheMode.*;
import static org.apache.ignite.cache.CacheRebalanceMode.*;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;

/**
 * Test for distributed queries with replicated client cache and node restarts.
 */
public class IgniteCacheClientQueryReplicatedNodeRestartSelfTest extends GridCommonAbstractTest {
    /** */
    private static final String QRY = "select co.id, count(*) cnt\n" +
        "from \"pe\".Person pe, \"pr\".Product pr, \"co\".Company co, \"pu\".Purchase pu\n" +
        "where pe.id = pu.personId and pu.productId = pr.id and pr.companyId = co.id \n" +
        "group by co.id order by cnt desc, co.id";

    /** */
    private static final P1<ClusterNode> DATA_NODES_FILTER = new P1<ClusterNode>() {
            @Override public boolean apply(ClusterNode clusterNode) {
                String gridName = clusterNode.attribute(IgniteNodeAttributes.ATTR_GRID_NAME);

                return !gridName.endsWith(String.valueOf(GRID_CNT - 1)); // The last one is client only.
            }
        };

    /** */
    private static final List<List<?>> FAKE = new LinkedList<>();

    /** */
    private static final int GRID_CNT = 5;

    /** */
    private static final int PERS_CNT = 600;

    /** */
    private static final int PURCHASE_CNT = 6000;

    /** */
    private static final int COMPANY_CNT = 25;

    /** */
    private static final int PRODUCT_CNT = 100;

    /** */
    private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);

    /** {@inheritDoc} */
    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
        X.println("grid name: " + gridName);

        IgniteConfiguration c = super.getConfiguration(gridName);

        TcpDiscoverySpi disco = new TcpDiscoverySpi();

        disco.setIpFinder(ipFinder);

        c.setDiscoverySpi(disco);

        int i = 0;

        CacheConfiguration<?, ?>[] ccs = new CacheConfiguration[4];

        for (String name : F.asList("co", "pr", "pe", "pu")) {
            CacheConfiguration<?, ?> cc = defaultCacheConfiguration();

            cc.setNodeFilter(DATA_NODES_FILTER);
            cc.setName(name);
            cc.setCacheMode(REPLICATED);
            cc.setWriteSynchronizationMode(FULL_SYNC);
            cc.setAtomicityMode(TRANSACTIONAL);
            cc.setRebalanceMode(SYNC);
            cc.setAffinity(new RendezvousAffinityFunction(false, 50));

            switch (name) {
                case "co":
                    cc.setIndexedTypes(
                        Integer.class, Company.class
                    );

                    break;

                case "pr":
                    cc.setIndexedTypes(
                        Integer.class, Product.class
                    );

                    break;

                case "pe":
                    cc.setIndexedTypes(
                        Integer.class, Person.class
                    );

                    break;

                case "pu":
                    cc.setIndexedTypes(
                        AffinityKey.class, Purchase.class
                    );

                    break;
            }

            ccs[i++] = cc;
        }

        c.setCacheConfiguration(ccs);

        return c;
    }

    /**
     *
     */
    private void fillCaches() {
        IgniteCache<Integer, Company> co = grid(0).cache("co");

        for (int i = 0; i < COMPANY_CNT; i++)
            co.put(i, new Company(i));

        IgniteCache<Integer, Product> pr = grid(0).cache("pr");

        Random rnd = new GridRandom();

        for (int i = 0; i < PRODUCT_CNT; i++)
            pr.put(i, new Product(i, rnd.nextInt(COMPANY_CNT)));

        IgniteCache<Integer, Person> pe = grid(0).cache("pe");

        for (int i = 0; i < PERS_CNT; i++)
            pe.put(i, new Person(i));

        IgniteCache<AffinityKey<Integer>, Purchase> pu = grid(0).cache("pu");

        for (int i = 0; i < PURCHASE_CNT; i++) {
            int persId = rnd.nextInt(PERS_CNT);
            int prodId = rnd.nextInt(PRODUCT_CNT);

            pu.put(new AffinityKey<>(i, persId), new Purchase(persId, prodId));
        }
    }

    /**
     * @param c Cache.
     * @param client If it must be a client cache.
     */
    private void assertClient(IgniteCache<?,?> c, boolean client) {
        assertTrue(((IgniteCacheProxy)c).context().affinityNode() == !client);
    }

    /**
     * @throws Exception If failed.
     */
    public void testRestarts() throws Exception {
        int duration = 90 * 1000;
        int qryThreadNum = 5;
        int restartThreadsNum = 2; // 2 of 4 data nodes
{'fraction_non_alphanumeric': 0.11740171468293813, 'fraction_numerical': 0.006951417316752916, 'mean_word_length': 2.303904057157438, 'pattern_counts': {'":': 8, '<': 39, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '540763', 'n_tokens_mistral': 3837, 'n_tokens_neox': 3604, 'n_words': 967}