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

import java.util.AbstractMap.SimpleEntry;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import javax.cache.expiry.AccessedExpiryPolicy;
import javax.cache.expiry.CreatedExpiryPolicy;
import javax.cache.expiry.Duration;
import javax.cache.expiry.ModifiedExpiryPolicy;
import com.google.common.collect.ImmutableSet;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheKeyConfiguration;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.cache.PartitionLossPolicy;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.ClientConfiguration;
import org.apache.ignite.configuration.ClientConnectorConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.binary.AbstractBinaryArraysTest;
import org.apache.ignite.internal.client.thin.ClientServerError;
import org.apache.ignite.internal.processors.cache.CacheEnumOperationsAbstractTest.TestEnum;
import org.apache.ignite.internal.processors.odbc.ClientListenerProcessor;
import org.apache.ignite.internal.processors.platform.cache.expiry.PlatformExpiryPolicy;
import org.apache.ignite.internal.processors.platform.client.ClientStatus;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.mxbean.ClientProcessorMXBean;
import org.apache.ignite.spi.systemview.view.SystemView;
import org.apache.ignite.spi.systemview.view.TransactionView;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionIsolation;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;

import static org.apache.ignite.internal.processors.cache.CacheEnumOperationsAbstractTest.TestEnum.VAL1;
import static org.apache.ignite.internal.processors.cache.CacheEnumOperationsAbstractTest.TestEnum.VAL2;
import static org.apache.ignite.internal.processors.cache.CacheEnumOperationsAbstractTest.TestEnum.VAL3;
import static org.apache.ignite.internal.processors.cache.transactions.IgniteTxManager.TXS_MON_LIST;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.runAsync;
import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE;
import static org.junit.Assert.assertArrayEquals;

/**
 * Thin client functional tests.
 */
public class FunctionalTest extends AbstractBinaryArraysTest {
    /** Per test timeout */
    @SuppressWarnings("deprecation")
    @Rule
    public Timeout globalTimeout = new Timeout((int)GridTestUtils.DFLT_TEST_TIMEOUT);

    /**
     * Tested API:
     * <ul>
     * <li>{@link IgniteClient#cache(String)}</li>
     * <li>{@link IgniteClient#getOrCreateCache(ClientCacheConfiguration)}</li>
     * <li>{@link IgniteClient#cacheNames()}</li>
     * <li>{@link IgniteClient#createCache(String)}</li>
     * <li>{@link IgniteClient#createCache(ClientCacheConfiguration)}</li>
     * <li>{@link IgniteCache#size(CachePeekMode...)}</li>
     * </ul>
     */
    @Test
    public void testCacheManagement() throws Exception {
        try (LocalIgniteCluster ignored = LocalIgniteCluster.start(2);
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            final String CACHE_NAME = "testCacheManagement";

            ClientCacheConfiguration cacheCfg = new ClientCacheConfiguration().setName(CACHE_NAME)
                .setCacheMode(CacheMode.REPLICATED)
                .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);

            int key = 1;
            Person val = new Person(key, Integer.toString(key));

            ClientCache<Integer, Person> cache = client.getOrCreateCache(cacheCfg);

            cache.put(key, val);

            assertEquals(1, cache.size());
            assertEquals(2, cache.size(CachePeekMode.ALL));

            cache = client.cache(CACHE_NAME);

            Person cachedVal = cache.get(key);

            assertEquals(val, cachedVal);

            Object[] cacheNames = new TreeSet<>(client.cacheNames()).toArray();

            assertArrayEquals(new TreeSet<>(Arrays.asList(Config.DEFAULT_CACHE_NAME, CACHE_NAME)).toArray(), cacheNames);

            client.destroyCache(CACHE_NAME);

            cacheNames = client.cacheNames().toArray();

            assertArrayEquals(new Object[] {Config.DEFAULT_CACHE_NAME}, cacheNames);

            cache = client.createCache(CACHE_NAME);

            assertFalse(cache.containsKey(key));

            cacheNames = client.cacheNames().toArray();

            assertArrayEquals(new TreeSet<>(Arrays.asList(Config.DEFAULT_CACHE_NAME, CACHE_NAME)).toArray(), cacheNames);

            client.destroyCache(CACHE_NAME);

            cache = client.createCache(cacheCfg);

            assertFalse(cache.containsKey(key));

            assertArrayEquals(new TreeSet<>(Arrays.asList(Config.DEFAULT_CACHE_NAME, CACHE_NAME)).toArray(), cacheNames);
        }
    }

    /**
     * Tested API:
     * <ul>
     * <li>{@link ClientCache#getName()}</li>
     * <li>{@link ClientCache#getConfiguration()}</li>
     * </ul>
     */
    @Test
    public void testCacheConfiguration() throws Exception {
        final String dataRegionName = "functional-test-data-region";

        IgniteConfiguration cfg = Config.getServerConfiguration()
            .setDataStorageConfiguration(new DataStorageConfiguration()
                .setDefaultDataRegionConfiguration(new DataRegionConfiguration()
                    .setName(dataRegionName)));

        try (Ignite ignored = Ignition.start(cfg);
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            final String CACHE_NAME = "testCacheConfiguration";

            ClientCacheConfiguration cacheCfgTemplate = new ClientCacheConfiguration().setName(CACHE_NAME)
                .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
                .setBackups(3)
                .setCacheMode(CacheMode.PARTITIONED)
                .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
                .setEagerTtl(false)
                .setGroupName("FunctionalTest")
                .setDefaultLockTimeout(12345)
                .setPartitionLossPolicy(PartitionLossPolicy.READ_WRITE_SAFE)
                .setReadFromBackup(true)
                .setRebalanceBatchSize(67890)
                .setRebalanceBatchesPrefetchCount(102938)
                .setRebalanceDelay(54321)
                .setRebalanceMode(CacheRebalanceMode.SYNC)
                .setRebalanceOrder(2)
                .setRebalanceThrottle(564738)
                .setRebalanceTimeout(142536)
                .setKeyConfiguration(new CacheKeyConfiguration("Employee", "orgId"))
                .setQueryEntities(new QueryEntity(int.class.getName(), "Employee")
                    .setTableName("EMPLOYEE")
                    .setFields(
                        Stream.of(
                            new SimpleEntry<>("id", Integer.class.getName()),
                            new SimpleEntry<>("orgId", Integer.class.getName())
                        ).collect(Collectors.toMap(
                            SimpleEntry::getKey, SimpleEntry::getValue, (a, b) -> a, LinkedHashMap::new
                        ))
                    )
                    // During query normalization null keyFields become empty set.
                    // Set empty collection for comparator.
                    .setKeyFields(Collections.emptySet())
                    .setKeyFieldName("id")
                    .setNotNullFields(Collections.singleton("id"))
                    .setDefaultFieldValues(Collections.singletonMap("id", 0))
                    .setIndexes(Collections.singletonList(new QueryIndex("id", true, "IDX_EMPLOYEE_ID")))
                    .setAliases(Stream.of("id", "orgId").collect(Collectors.toMap(f -> f, String::toUpperCase)))
                )
                .setExpiryPolicy(new PlatformExpiryPolicy(10, 20, 30))
                .setCopyOnRead(!CacheConfiguration.DFLT_COPY_ON_READ)
                .setDataRegionName(dataRegionName)
                .setMaxConcurrentAsyncOperations(4)
                .setMaxQueryIteratorsCount(4)
                .setOnheapCacheEnabled(true)
                .setQueryDetailMetricsSize(1024)
                .setQueryParallelism(4)
                .setSqlEscapeAll(true)
                .setSqlIndexMaxInlineSize(1024)
                .setSqlSchema("functional-test-schema")
                .setStatisticsEnabled(true);

            ClientCacheConfiguration cacheCfg = new ClientCacheConfiguration(cacheCfgTemplate);

            ClientCache<Object, Object> cache = client.createCache(cacheCfg);

            assertEquals(CACHE_NAME, cache.getName());

            assertTrue(Comparers.equal(cacheCfgTemplate, cache.getConfiguration()));
        }
    }

    /**
     * Tested API:
     * <ul>
     * <li>{@link Ignition#startClient(ClientConfiguration)}</li>
     * <li>{@link IgniteClient#getOrCreateCache(String)}</li>
     * <li>{@link ClientCache#put(Object, Object)}</li>
     * <li>{@link ClientCache#get(Object)}</li>
     * <li>{@link ClientCache#containsKey(Object)}</li>
     * <li>{@link ClientCache#clear(Object)}</li>
     * </ul>
     */
    @Test
    public void testPutGet() throws Exception {
        // Existing cache, primitive key and object value
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, Person> cache = client.getOrCreateCache(Config.DEFAULT_CACHE_NAME);

            Integer key = 1;
            Person val = new Person(key, "Joe");

            cache.put(key, val);

            assertTrue(cache.containsKey(key));

            Person cachedVal = cache.get(key);

            assertEquals(val, cachedVal);

            cache.clear(key);

            assertFalse(cache.containsKey(key));

            assertNull(cache.get(key));
        }

        // Non-existing cache, object key and primitive value
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Person, Integer> cache = client.getOrCreateCache("testPutGet");

            Integer val = 1;

            Person key = new Person(val, "Joe");

            cache.put(key, val);

            Integer cachedVal = cache.get(key);

            assertEquals(val, cachedVal);

            cache.clear(key);

            assertFalse(cache.containsKey(key));

            assertNull(cache.get(key));
        }

        // Object key and Object value
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Person, Person> cache = client.getOrCreateCache("testPutGet");

            Person key = new Person(1, "Joe Key");

            Person val = new Person(1, "Joe Value");

            cache.put(key, val);

            Person cachedVal = cache.get(key);

            assertEquals(val, cachedVal);

            cache.clear(key);

            assertFalse(cache.containsKey(key));

            assertNull(cache.get(key));
        }
    }

    /**
     * Test cache operations with different data types.
     */
    @Test
    public void testDataTypes() throws Exception {
        try (Ignite ignite = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ignite.getOrCreateCache(Config.DEFAULT_CACHE_NAME);

            Person person = new Person(1, "name");

            // Primitive and built-in types.
            checkDataType(client, ignite, (byte)1);
            checkDataType(client, ignite, (short)1);
            checkDataType(client, ignite, 1);
            checkDataType(client, ignite, 1L);
            checkDataType(client, ignite, 1.0f);
            checkDataType(client, ignite, 1.0d);
            checkDataType(client, ignite, 'c');
            checkDataType(client, ignite, true);
            checkDataType(client, ignite, "string");
            checkDataType(client, ignite, UUID.randomUUID());
            checkDataType(client, ignite, new Date());

            // Enum.
            checkDataType(client, ignite, VAL1);

            // Binary object.
            checkDataType(client, ignite, person);

            // Arrays.
            checkDataType(client, ignite, new byte[] {(byte)1});
            checkDataType(client, ignite, new short[] {(short)1});
            checkDataType(client, ignite, new int[] {1});
            checkDataType(client, ignite, new long[] {1L});
            checkDataType(client, ignite, new float[] {1.0f});
            checkDataType(client, ignite, new double[] {1.0d});
            checkDataType(client, ignite, new char[] {'c'});
            checkDataType(client, ignite, new boolean[] {true});
            checkDataType(client, ignite, new String[] {"string"});
            checkDataType(client, ignite, new UUID[] {UUID.randomUUID()});
            checkDataType(client, ignite, new Date[] {new Date()});
            checkDataType(client, ignite, new int[][] {new int[] {1}});

            checkDataType(client, ignite, new TestEnum[] {VAL1, VAL2, VAL3});

            checkDataType(client, ignite, new Person[] {person});
            checkDataType(client, ignite, new Person[][] {new Person[] {person}});
            checkDataType(client, ignite, new Object[] {1, "string", person, new Person[] {person}});

            // Lists.
            checkDataType(client, ignite, Collections.emptyList());
            checkDataType(client, ignite, Collections.singletonList(person));
            checkDataType(client, ignite, Arrays.asList(person, person));
            checkDataType(client, ignite, new ArrayList<>(Arrays.asList(person, person)));
            checkDataType(client, ignite, new LinkedList<>(Arrays.asList(person, person)));
            checkDataType(client, ignite, Arrays.asList(Arrays.asList(person, person), person));

            // Sets.
            checkDataType(client, ignite, Collections.emptySet());
            checkDataType(client, ignite, Collections.singleton(person));
            checkDataType(client, ignite, new HashSet<>(Arrays.asList(1, 2)));
            checkDataType(client, ignite, new HashSet<>(Arrays.asList(Arrays.asList(person, person), person)));
            checkDataType(client, ignite, new HashSet<>(new ArrayList<>(Arrays.asList(Arrays.asList(person,
                person), person))));

            // Maps.
            checkDataType(client, ignite, Collections.emptyMap());
            checkDataType(client, ignite, Collections.singletonMap(1, person));
            checkDataType(client, ignite, F.asMap(1, person));
            checkDataType(client, ignite, new HashMap<>(F.asMap(1, person)));
            checkDataType(client, ignite, new HashMap<>(F.asMap(new HashSet<>(Arrays.asList(1, 2)),
                Arrays.asList(person, person))));
        }
    }

    /**
     * Check that we get the same value from the cache as we put before.
     *
     * @param client Thin client.
     * @param ignite Ignite node.
     * @param obj Value of data type to check.
     */
    private void checkDataType(IgniteClient client, Ignite ignite, Object obj) {
        IgniteCache<Object, Object> thickCache = ignite.cache(Config.DEFAULT_CACHE_NAME);
        ClientCache<Object, Object> thinCache = client.cache(Config.DEFAULT_CACHE_NAME);

        Integer key = 1;

        thinCache.put(key, obj);

        assertTrue(thinCache.containsKey(key));

        Object cachedObj = thinCache.get(key);

        assertEqualsArraysAware(obj, cachedObj);

        assertEqualsArraysAware(obj, thickCache.get(key));

        assertEquals(client.binary().typeId(obj.getClass().getName()), ignite.binary().typeId(obj.getClass().getName()));

        // Server-side comparison with the original object.
        assertTrue(thinCache.replace(key, obj, obj));

        // Server-side comparison with the restored object.
        assertTrue(thinCache.remove(key, cachedObj));
    }

    /**
     * Tested API:
     * <ul>
     * <li>{@link ClientCache#putAll(Map)}</li>
     * <li>{@link ClientCache#getAll(Set)}</li>
     * <li>{@link ClientCache#containsKeys(Set)} (Set)}</li>
     * <li>{@link ClientCache#clear()}</li>
     * <li>{@link ClientCache#clearAll(Set)} ()}</li>
     * </ul>
     */
    @Test
    public void testBatchPutGet() throws Exception {
        // Existing cache, primitive key and object value
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, Person> cache = client.cache(Config.DEFAULT_CACHE_NAME);

            Map<Integer, Person> data = IntStream
                .rangeClosed(1, 1000).boxed()
                .collect(Collectors.toMap(i -> i, i -> new Person(i, String.format("Person %s", i))));

            assertFalse(cache.containsKeys(data.keySet()));

            cache.putAll(data);

            assertTrue(cache.containsKeys(data.keySet()));

            Map<Integer, Person> cachedData = cache.getAll(data.keySet());

            assertEquals(data, cachedData);
        }

        // Non-existing cache, object key and primitive value
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Person, Integer> cache = client.createCache("testBatchPutGet");

            Map<Person, Integer> data = IntStream
                .rangeClosed(1, 1000).boxed()
                .collect(Collectors.toMap(i -> new Person(i, String.format("Person %s", i)), i -> i));

            assertFalse(cache.containsKeys(data.keySet()));

            cache.putAll(data);

            assertTrue(cache.containsKeys(data.keySet()));

            Map<Person, Integer> cachedData = cache.getAll(data.keySet());

            assertEquals(data, cachedData);

            Set<Person> clearKeys = new HashSet<>();

            Iterator<Person> keyIter = data.keySet().iterator();

            for (int i = 0; i < 100; i++)
                clearKeys.add(keyIter.next());

            cache.clearAll(clearKeys);

            assertFalse(cache.containsKeys(clearKeys));
            assertTrue(cache.containsKeys(
                data.keySet().stream().filter(key -> !data.containsKey(key)).collect(Collectors.toSet())));
            assertEquals(data.size() - clearKeys.size(), cache.size(CachePeekMode.ALL));

            cache.clear();

            assertFalse(cache.containsKeys(data.keySet()));
            assertEquals(0, cache.size(CachePeekMode.ALL));
        }
    }

    /**
     * Tested API:
     * <ul>
     * <li>{@link ClientCache#getAndPut(Object, Object)}</li>
     * <li>{@link ClientCache#getAndRemove(Object)}</li>
     * <li>{@link ClientCache#getAndReplace(Object, Object)}</li>
     * <li>{@link ClientCache#putIfAbsent(Object, Object)}</li>
     * <li>{@link ClientCache#getAndPutIfAbsent(Object, Object)}</li>
     * </ul>
     */
    @Test
    public void testAtomicPutGet() throws Exception {
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache("testRemoveReplace");

            assertNull(cache.getAndPut(1, "1"));
            assertEquals("1", cache.getAndPut(1, "1.1"));

            assertEquals("1.1", cache.getAndRemove(1));
            assertNull(cache.getAndRemove(1));

            assertTrue(cache.putIfAbsent(1, "1"));
            assertFalse(cache.putIfAbsent(1, "1.1"));

            assertEquals("1", cache.getAndReplace(1, "1.1"));
            assertEquals("1.1", cache.getAndReplace(1, "1"));
            assertNull(cache.getAndReplace(2, "2"));

            assertEquals("1", cache.getAndPutIfAbsent(1, "1.1"));
            assertEquals("1", cache.get(1));
            assertNull(cache.getAndPutIfAbsent(3, "3"));
            assertEquals("3", cache.get(3));
        }
    }

    /**
     * Tested API:
     * <ul>
     * <li>{@link ClientCache#replace(Object, Object)}</li>
     * <li>{@link ClientCache#replace(Object, Object, Object)}</li>
     * <li>{@link ClientCache#remove(Object)}</li>
     * <li>{@link ClientCache#remove(Object, Object)}</li>
     * <li>{@link ClientCache#removeAll()}</li>
     * <li>{@link ClientCache#removeAll(Set)}</li>
     * </ul>
     */
    @Test
    public void testRemoveReplace() throws Exception {
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache("testRemoveReplace");

            Map<Integer, String> data = IntStream.rangeClosed(1, 100).boxed()
                .collect(Collectors.toMap(i -> i, Object::toString));

            cache.putAll(data);

            assertFalse(cache.replace(1, "2", "3"));
            assertEquals("1", cache.get(1));
            assertTrue(cache.replace(1, "1", "3"));
            assertEquals("3", cache.get(1));

            assertFalse(cache.replace(101, "101"));
            assertNull(cache.get(101));
            assertTrue(cache.replace(100, "101"));
            assertEquals("101", cache.get(100));

            assertFalse(cache.remove(101));
            assertTrue(cache.remove(100));
            assertNull(cache.get(100));

            assertFalse(cache.remove(99, "100"));
            assertEquals("99", cache.get(99));
            assertTrue(cache.remove(99, "99"));
            assertNull(cache.get(99));

            cache.put(101, "101");

            cache.removeAll(data.keySet());
            assertEquals(1, cache.size());
            assertEquals("101", cache.get(101));

            cache.removeAll();
            assertEquals(0, cache.size());
        }
    }

    /**
     * Test client fails on start if server is unavailable
     */
    @Test
    public void testClientFailsOnStart() {
        ClientConnectionException expEx = null;

        //noinspection EmptyTryBlock
        try (IgniteClient ignored = Ignition.startClient(getClientConfiguration())) {
            // No-op.
        }
        catch (ClientConnectionException connEx) {
            expEx = connEx;
        }
        catch (Exception ex) {
            fail(String.format(
                "%s expected but %s was received: %s",
                ClientConnectionException.class.getName(),
                ex.getClass().getName(),
                ex
            ));
        }

        assertNotNull(
            String.format("%s expected but no exception was received", ClientConnectionException.class.getName()),
            expEx
        );
    }


    /**
     * Test PESSIMISTIC REPEATABLE_READ tx holds lock and other tx should be timed out.
     */
    @Test
    public void testPessimisticRepeatableReadsTransactionHoldsLock() throws Exception {
        testPessimisticTxLocking(REPEATABLE_READ);
    }

    /**
     * Test PESSIMISTIC SERIALIZABLE tx holds lock and other tx should be timed out.
     */
    @Test
    public void testPessimisticSerializableTransactionHoldsLock() throws Exception {
        testPessimisticTxLocking(SERIALIZABLE);
    }

    /**
     * Test pessimistic tx holds the lock.
     */
    private void testPessimisticTxLocking(TransactionIsolation isolation) throws Exception {
        try (Ignite ignite = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache(new ClientCacheConfiguration()
                    .setName("cache")
                    .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );
            cache.put(0, "value0");

            IgniteInternalFuture<?> fut;

            try (ClientTransaction tx = client.transactions().txStart(PESSIMISTIC, isolation)) {
                assertEquals("value0", cache.get(0));

                CyclicBarrier barrier = new CyclicBarrier(2);

                fut = GridTestUtils.runAsync(() -> {
                    try (ClientTransaction tx2 = client.transactions().txStart(OPTIMISTIC, REPEATABLE_READ, 500)) {
                        cache.put(0, "value2");
                        tx2.commit();
                    }
                    finally {
                        try {
                            barrier.await(2000, TimeUnit.MILLISECONDS);
                        }
                        catch (Throwable ignore) {
                            // No-op.
                        }
                    }
                });

                barrier.await(2000, TimeUnit.MILLISECONDS);

                tx.commit();
            }

            assertEquals("value0", cache.get(0));

            assertThrowsAnyCause(null, fut::get, ClientException.class,
                    "Failed to acquire lock within provided timeout");
        }
    }

    /**
     * Test OPTIMISTIC SERIALIZABLE tx rolls backs if another TX commits.
     */
    @Test
    public void testOptimitsticSerializableTransactionHoldsLock() throws Exception {
        try (Ignite ignite = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache(new ClientCacheConfiguration()
                    .setName("cache")
                    .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );
            cache.put(0, "value0");

            final CountDownLatch latch = new CountDownLatch(1);

            try (ClientTransaction tx = client.transactions().txStart(OPTIMISTIC, SERIALIZABLE)) {
                assertEquals("value0", cache.get(0));

                IgniteInternalFuture<?> fut = GridTestUtils.runAsync(() -> {
                    try (ClientTransaction tx2 = client.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
                        cache.put(0, "value2");
                        tx2.commit();
                    }
                    finally {
                        latch.countDown();
                    }
                });

                latch.await();

                cache.put(0, "value1");

                fut.get();

                assertThrowsAnyCause(null, () -> {
                    tx.commit();
                    return null;
                }, ClientException.class, "read/write conflict");
            }

            assertEquals("value2", cache.get(0));
        }
    }

    /**
     * Test OPTIMISTIC REPEATABLE_READ tx doesn't conflict with a regular cache put.
     */
    @Test
    public void testOptimitsticRepeatableReadUpdatesValue() throws Exception {
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache(new ClientCacheConfiguration()
                    .setName("cache")
                    .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );
            cache.put(0, "value0");

            try (ClientTransaction tx = client.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) {
                assertEquals("value0", cache.get(0));

                cache.put(0, "value1");

                GridTestUtils.runAsync(() -> {
                    assertEquals("value0", cache.get(0));

                    cache.put(0, "value2");

                    assertEquals("value2", cache.get(0));
                }).get();

                tx.commit();
            }

            assertEquals("value1", cache.get(0));
        }
    }

    /**
     * Test that client-connector worker can process further transactional requests (resume transactions) after
     * external termination of previous transaction.
     */
    @Test
    public void testTxResumeAfterTxTimeout() throws Exception {
        IgniteConfiguration cfg = Config.getServerConfiguration().setClientConnectorConfiguration(
            new ClientConnectorConfiguration().setThreadPoolSize(1));

        try (Ignite ignite = Ignition.start(cfg); IgniteClient client = Ignition.startClient(getClientConfiguration())) {
            String cacheName = "cache";

            IgniteCache<Object, Object> igniteCache = ignite.createCache(new CacheConfiguration<>(cacheName)
                .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));

            try (ClientTransaction clientTx = client.transactions().txStart()) {
                runAsync(() -> {
                    try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
                        igniteCache.put(0, 0); // Lock key by ignite node.

                        try {
                            // Start, but don't close the transaction (to keep it in the threadMap after timeout).
                            client.transactions().txStart(PESSIMISTIC, READ_COMMITTED, 200L);

                            // Wait until transaction interrupted externally by timeout.
                            client.cache(cacheName).put(0, 0);

                            fail();
                        }
                        catch (ClientException ignored) {
                            // Expected.
                        }
                    }
                }).get();

                // Resume tx in the worker with interrupted transaction.
                assertFalse(client.cache(cacheName).containsKey(0));
            }
        }
    }

    /**
     * Test transactions.
     */
    @Test
    public void testTransactions() throws Exception {
        try (Ignite ignite = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache(new ClientCacheConfiguration()
                .setName("cache")
                .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );

            cache.put(0, "value0");
            cache.put(1, "value1");

            // Test nested transactions is not possible.
            try (ClientTransaction tx = client.transactions().txStart()) {
                try (ClientTransaction tx1 = client.transactions().txStart()) {
                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }
            }

            // Test implicit rollback when transaction closed.
            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.put(1, "value2");
            }

            assertEquals("value1", cache.get(1));

            // Test explicit rollback.
            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.put(1, "value2");

                tx.rollback();
            }

            assertEquals("value1", cache.get(1));

            // Test commit.
            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.put(1, "value2");

                tx.commit();
            }

            assertEquals("value2", cache.get(1));

            // Test end of already completed transaction.
            ClientTransaction tx0 = client.transactions().txStart();

            tx0.close();

            try {
                tx0.commit();

                fail();
            }
            catch (ClientException expected) {
                // No-op.
            }

            // Test end of outdated transaction.
            try (ClientTransaction tx = client.transactions().txStart()) {
                try {
                    tx0.commit();

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                tx.commit();
            }

            // Test transaction with a timeout.
            long TX_TIMEOUT = 200L;

            try (ClientTransaction tx = client.transactions().txStart(PESSIMISTIC, READ_COMMITTED, TX_TIMEOUT)) {
                long txStartedTime = U.currentTimeMillis();

                cache.put(1, "value3");

                while (txStartedTime + TX_TIMEOUT >= U.currentTimeMillis())
                    U.sleep(100L);

                try {
                    cache.put(1, "value4");

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                try {
                    tx.commit();

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }
            }

            assertEquals("value2", cache.get(1));

            cache.put(1, "value5");

            // Test failover.
            ClientProcessorMXBean mxBean = getMxBean(ignite.name(), "Clients",
                ClientListenerProcessor.class, ClientProcessorMXBean.class);

            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.put(1, "value6");

                mxBean.dropAllConnections();

                try {
                    cache.put(1, "value7");

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                // Start new transaction doesn't recover cache operations on failed channel.
                try (ClientTransaction tx1 = client.transactions().txStart()) {
                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                try {
                    cache.get(1);

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                // Close outdated transaction doesn't recover cache operations on failed channel.
                tx0.close();

                try {
                    cache.get(1);

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }
            }

            assertEquals("value5", cache.get(1));

            // Test concurrent transactions in different connections.
            try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) {
                ClientCache<Integer, String> cache1 = client1.cache("cache");

                try (ClientTransaction tx = client.transactions().txStart(OPTIMISTIC, READ_COMMITTED)) {
                    cache.put(0, "value8");

                    try (ClientTransaction tx1 = client1.transactions().txStart(OPTIMISTIC, READ_COMMITTED)) {
                        assertEquals("value8", cache.get(0));
                        assertEquals("value0", cache1.get(0));

                        cache1.put(1, "value9");

                        assertEquals("value5", cache.get(1));
                        assertEquals("value9", cache1.get(1));

                        tx1.commit();

                        assertEquals("value9", cache.get(1));
                    }

                    assertEquals("value0", cache1.get(0));

                    tx.commit();

                    assertEquals("value8", cache1.get(0));
                }
            }

            // Check different types of cache operations.
            try (ClientTransaction tx = client.transactions().txStart()) {
                // Operations: put, putAll, putIfAbsent.
                cache.put(2, "value10");
                cache.putAll(F.asMap(1, "value11", 3, "value12"));
                cache.putIfAbsent(4, "value13");

                // Operations: get, getAll, getAndPut, getAndRemove, getAndReplace, getAndPutIfAbsent.
                assertEquals("value10", cache.get(2));
                assertEquals(F.asMap(1, "value11", 2, "value10"),
                    cache.getAll(new HashSet<>(Arrays.asList(1, 2))));
                assertEquals("value13", cache.getAndPut(4, "value14"));
                assertEquals("value14", cache.getAndPutIfAbsent(4, "valueDiscarded"));
                assertEquals("value14", cache.get(4));
                assertEquals("value14", cache.getAndReplace(4, "value15"));
                assertEquals("value15", cache.getAndRemove(4));
                assertNull(cache.getAndPutIfAbsent(10, "valuePutIfAbsent"));
                assertEquals("valuePutIfAbsent", cache.get(10));

                // Operations: contains.
                assertTrue(cache.containsKey(2));
                assertFalse(cache.containsKey(4));
                assertTrue(cache.containsKeys(ImmutableSet.of(2, 10)));
                assertFalse(cache.containsKeys(ImmutableSet.of(2, 4)));

                // Operations: replace.
                cache.put(4, "");
                assertTrue(cache.replace(4, "value16"));
                assertTrue(cache.replace(4, "value16", "value17"));

                // Operations: remove, removeAll
                cache.putAll(F.asMap(5, "", 6, ""));
                assertTrue(cache.remove(5));
                assertTrue(cache.remove(4, "value17"));
                cache.removeAll(new HashSet<>(Arrays.asList(3, 6)));
                assertFalse(cache.containsKey(3));
                assertFalse(cache.containsKey(6));

                tx.rollback();
            }

            assertEquals(F.asMap(0, "value8", 1, "value9"),
                cache.getAll(new HashSet<>(Arrays.asList(0, 1))));
            assertFalse(cache.containsKey(2));

            // Test concurrent transactions started by different threads.
            try (ClientTransaction tx = client.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
                CyclicBarrier barrier = new CyclicBarrier(2);

                cache.put(0, "value18");

                IgniteInternalFuture<?> fut = GridTestUtils.runAsync(() -> {
                    try (ClientTransaction tx1 = client.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
                        cache.put(1, "value19");

                        barrier.await();

                        assertEquals("value8", cache.get(0));

                        barrier.await();

                        tx1.commit();

                        barrier.await();

                        assertEquals("value18", cache.get(0));
                    }
                    catch (InterruptedException | BrokenBarrierException ignore) {
                        // No-op.
                    }
                });

                barrier.await();

                assertEquals("value9", cache.get(1));

                barrier.await();

                tx.commit();

                barrier.await();

                assertEquals("value19", cache.get(1));

                fut.get();
            }

            // Test transaction usage by different threads.
            try (ClientTransaction tx = client.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
                cache.put(0, "value20");

                GridTestUtils.runAsync(() -> {
                    // Implicit transaction started here.
                    cache.put(1, "value21");

                    assertEquals("value18", cache.get(0));

                    try {
                        // Transaction can't be commited by another thread.
                        tx.commit();

                        fail();
                    }
                    catch (ClientException expected) {
                        // No-op.
                    }

                    // Transaction can be closed by another thread.
                    tx.close();

                    assertEquals("value18", cache.get(0));
                }).get();

                assertEquals("value21", cache.get(1));

                try {
                    // Transaction can't be commited after another thread close this transaction.
                    tx.commit();

                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                assertEquals("value18", cache.get(0));

                // Start implicit transaction after explicit transaction has been closed by another thread.
                cache.put(0, "value22");

                GridTestUtils.runAsync(() -> assertEquals("value22", cache.get(0))).get();

                // New explicit transaction can be started after current transaction has been closed by another thread.
                try (ClientTransaction tx1 = client.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
                    cache.put(0, "value23");

                    tx1.commit();
                }

                assertEquals("value23", cache.get(0));
            }

            // Test active transactions limit.
            int txLimit = ignite.configuration().getClientConnectorConfiguration().getThinClientConfiguration()
                .getMaxActiveTxPerConnection();

            List<ClientTransaction> txs = new ArrayList<>(txLimit);

            for (int i = 0; i < txLimit; i++) {
                Thread t = new Thread(() -> txs.add(client.transactions().txStart()));

                t.start();

                t.join();
            }

            try (ClientTransaction ignored = client.transactions().txStart()) {
                fail();
            }
            catch (ClientException e) {
                ClientServerError cause = (ClientServerError)e.getCause();
                assertEquals(ClientStatus.TX_LIMIT_EXCEEDED, cause.getCode());
            }

            for (ClientTransaction tx : txs)
                tx.close();

            // Test that new transaction can be started after commit of the previous one without closing.
            ClientTransaction tx = client.transactions().txStart();
            tx.commit();

            tx = client.transactions().txStart();
            tx.rollback();

            // Test that new transaction can be started after rollback of the previous one without closing.
            tx = client.transactions().txStart();
            tx.commit();

            // Test that implicit transaction started after commit of previous one without closing.
            cache.put(0, "value24");

            GridTestUtils.runAsync(() -> assertEquals("value24", cache.get(0))).get();
        }
    }

    /**
     * Test transactions.
     */
    @Test
    public void testTransactionsAsync() throws Exception {
        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache(new ClientCacheConfiguration()
                    .setName("cache")
                    .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );

            cache.put(0, "value0");
            cache.put(1, "value1");

            // Test implicit rollback when transaction closed.
            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.putAsync(1, "value2").get();
            }

            assertEquals("value1", cache.get(1));

            // Test explicit rollback.
            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.putAsync(1, "value2").get();

                tx.rollback();
            }

            assertEquals("value1", cache.get(1));

            // Test commit.
            try (ClientTransaction tx = client.transactions().txStart()) {
                cache.putAsync(1, "value2").get();

                tx.commit();
            }

            assertEquals("value2", cache.get(1));
        }
    }

    /**
     * Test transactions with label.
     */
    @Test
    public void testTransactionsWithLabel() throws Exception {
        try (IgniteEx ignite = (IgniteEx)Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, String> cache = client.createCache(new ClientCacheConfiguration()
                .setName("cache")
                .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );

            SystemView<TransactionView> txsView = ignite.context().systemView().view(TXS_MON_LIST);

            cache.put(0, "value1");

            try (ClientTransaction tx = client.transactions().withLabel("label").txStart()) {
                cache.put(0, "value2");

                assertEquals(1, F.size(txsView.iterator()));

                TransactionView txv = txsView.iterator().next();

                assertEquals("label", txv.label());

                assertEquals("value2", cache.get(0));
            }

            assertEquals("value1", cache.get(0));

            try (ClientTransaction tx = client.transactions().withLabel("label1").withLabel("label2").txStart()) {
                cache.put(0, "value2");

                assertEquals(1, F.size(txsView.iterator()));

                TransactionView txv = txsView.iterator().next();

                assertEquals("label2", txv.label());

                tx.commit();
            }

            assertEquals("value2", cache.get(0));

            // Test concurrent with label and without label transactions.
            try (ClientTransaction tx = client.transactions().withLabel("label").txStart(PESSIMISTIC, READ_COMMITTED)) {
                CyclicBarrier barrier = new CyclicBarrier(2);

                cache.put(0, "value3");

                IgniteInternalFuture<?> fut = GridTestUtils.runAsync(() -> {
                    try (ClientTransaction tx1 = client.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
                        cache.put(1, "value3");

                        barrier.await();

                        assertEquals("value2", cache.get(0));

                        barrier.await();
                    }
                    catch (InterruptedException | BrokenBarrierException ignore) {
                        // No-op.
                    }
                });

                barrier.await();

                assertNull(cache.get(1));

                assertEquals(1, F.size(txsView.iterator(), txv -> txv.label() == null));
                assertEquals(1, F.size(txsView.iterator(), txv -> "label".equals(txv.label())));

                barrier.await();

                fut.get();
            }

            // Test nested transactions is not possible.
            try (ClientTransaction tx = client.transactions().withLabel("label1").txStart()) {
                try (ClientTransaction tx1 = client.transactions().txStart()) {
                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }

                try (ClientTransaction tx1 = client.transactions().withLabel("label2").txStart()) {
                    fail();
                }
                catch (ClientException expected) {
                    // No-op.
                }
            }
        }
    }

    /**
     * Test cache with expire policy.
     */
    @Test
    public void testExpirePolicy() throws Exception {
        long ttl = 600L;
        int MAX_RETRIES = 5;

        try (Ignite ignite = Ignition.start(Config.getServerConfiguration());
             IgniteClient client = Ignition.startClient(getClientConfiguration())
        ) {
            ClientCache<Integer, Object> cache = client.createCache(new ClientCacheConfiguration()
                .setName("cache")
                .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
            );

            Duration dur = new Duration(TimeUnit.MILLISECONDS, ttl);

            ClientCache<Integer, Object> cachePlcCreated = cache.withExpirePolicy(new CreatedExpiryPolicy(dur));
            ClientCache<Integer, Object> cachePlcUpdated = cache.withExpirePolicy(new ModifiedExpiryPolicy(dur));
            ClientCache<Integer, Object> cachePlcAccessed = cache.withExpirePolicy(new AccessedExpiryPolicy(dur));

            for (int i = 0; i < MAX_RETRIES; i++) {
                cache.clear();

                long ts = U.currentTimeMillis();

                cache.put(0, 0);
                cachePlcCreated.put(1, 1);
                cachePlcUpdated.put(2, 2);
                cachePlcAccessed.put(3, 3);

                U.sleep(ttl / 3 * 2);

                boolean containsKey0 = cache.containsKey(0);
                boolean containsKey1 = cache.containsKey(1);
                boolean containsKey2 = cache.containsKey(2);
                boolean containsKey3 = cache.containsKey(3);

                if (U.currentTimeMillis() - ts >= ttl) // Retry if this block execution takes too long.
                    continue;

                assertTrue(containsKey0);
                assertTrue(containsKey1);
                assertTrue(containsKey2);
                assertTrue(containsKey3);

                ts = U.currentTimeMillis();

                cachePlcCreated.put(1, 2);
                cachePlcCreated.get(1); // Update and access key with created expire policy.
                cachePlcUpdated.put(2, 3); // Update key with modified expire policy.
                cachePlcAccessed.get(3); // Access key with accessed expire policy.

                U.sleep(ttl / 3 * 2);

                containsKey0 = cache.containsKey(0);
                containsKey1 = cache.containsKey(1);
                containsKey2 = cache.containsKey(2);
                containsKey3 = cache.containsKey(3);

                if (U.currentTimeMillis() - ts >= ttl) // Retry if this block execution takes too long.
                    continue;

                assertTrue(containsKey0);
                assertFalse(containsKey1);
                assertTrue(containsKey2);
                assertTrue(containsKey3);

                U.sleep(ttl / 3 * 2);

                cachePlcUpdated.get(2); // Access key with updated expire policy.

                U.sleep(ttl / 3 * 2);

                assertTrue(cache.containsKey(0));
                assertFalse(cache.containsKey(1));
                assertFalse(cache.containsKey(2));
                assertFalse(cache.containsKey(3));

                // Expire policy, keep binary and transactional flags together.
                ClientCache<Integer, Object> binCache = cachePlcCreated.withKeepBinary();

                try (ClientTransaction tx = client.transactions().txStart()) {
                    binCache.put(4, new T2<>("test", "test"));

                    tx.commit();
                }

                assertTrue(binCache.get(4) instanceof BinaryObject);
                assertFalse(cache.get(4) instanceof BinaryObject);

                U.sleep(ttl / 3 * 4);

                assertFalse(cache.containsKey(4));

                return;
            }

            fail("Failed to check expire policy within " + MAX_RETRIES + " retries (block execution takes too long)");
        }
    }

    /** */
    private static ClientConfiguration getClientConfiguration() {
        return new ClientConfiguration()
            .setAddresses(Config.SERVER)
            .setSendBufferSize(0)
            .setReceiveBufferSize(0);
    }
}