| /* |
| * 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; |
| |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.UUID; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.CyclicBarrier; |
| import java.util.concurrent.ThreadLocalRandom; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.locks.Lock; |
| import javax.cache.CacheException; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteCache; |
| import org.apache.ignite.IgniteException; |
| import org.apache.ignite.IgniteLogger; |
| import org.apache.ignite.IgniteTransactions; |
| import org.apache.ignite.cache.CacheAtomicityMode; |
| import org.apache.ignite.cache.affinity.Affinity; |
| import org.apache.ignite.cache.affinity.AffinityFunction; |
| import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.configuration.NearCacheConfiguration; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.EventType; |
| import org.apache.ignite.internal.IgniteEx; |
| import org.apache.ignite.internal.IgniteInternalFuture; |
| import org.apache.ignite.internal.IgniteKernal; |
| import org.apache.ignite.internal.IgniteNodeAttributes; |
| import org.apache.ignite.internal.managers.communication.GridIoMessage; |
| import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; |
| import org.apache.ignite.internal.processors.affinity.GridAffinityFunctionContextImpl; |
| import org.apache.ignite.internal.processors.cache.GridCacheAdapter; |
| import org.apache.ignite.internal.processors.cache.GridCacheAffinityManager; |
| import org.apache.ignite.internal.processors.cache.GridCacheEntryEx; |
| import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicFullUpdateRequest; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheEntry; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest; |
| import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest; |
| import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; |
| import org.apache.ignite.internal.util.GridConcurrentHashSet; |
| import org.apache.ignite.internal.util.lang.GridAbsPredicate; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.G; |
| import org.apache.ignite.internal.util.typedef.PA; |
| import org.apache.ignite.internal.util.typedef.T2; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.lang.IgniteInClosure; |
| import org.apache.ignite.plugin.extensions.communication.Message; |
| import org.apache.ignite.resources.LoggerResource; |
| import org.apache.ignite.spi.IgniteSpiException; |
| import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; |
| import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; |
| import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode; |
| import org.apache.ignite.testframework.GridTestUtils; |
| import org.apache.ignite.testframework.MvccFeatureChecker; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.apache.ignite.transactions.Transaction; |
| import org.apache.ignite.transactions.TransactionConcurrency; |
| import org.apache.ignite.transactions.TransactionIsolation; |
| import org.jetbrains.annotations.Nullable; |
| import org.junit.Test; |
| |
| import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC; |
| import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; |
| import static org.apache.ignite.cache.CacheMode.PARTITIONED; |
| import static org.apache.ignite.cache.CacheRebalanceMode.SYNC; |
| import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; |
| import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC; |
| import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; |
| import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ; |
| import static org.apache.ignite.transactions.TransactionIsolation.SERIALIZABLE; |
| |
| /** |
| * |
| */ |
| public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstractTest { |
| /** */ |
| private CacheConfiguration ccfg; |
| |
| /** */ |
| private volatile CyclicBarrier updateBarrier; |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); |
| |
| cfg.setConsistentId(igniteInstanceName); |
| |
| ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true); |
| |
| TestCommunicationSpi commSpi = new TestCommunicationSpi(); |
| |
| commSpi.setSharedMemoryPort(-1); |
| |
| cfg.setCommunicationSpi(commSpi); |
| cfg.setCacheConfiguration(ccfg); |
| |
| return cfg; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void beforeTest() throws Exception { |
| MvccFeatureChecker.skipIfNotSupported(MvccFeatureChecker.Feature.ENTRY_LOCK); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void afterTest() throws Exception { |
| super.afterTest(); |
| |
| stopAllGrids(); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAtomicPutAllPrimaryMode() throws Exception { |
| atomicPut(true, null); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAtomicPutAllNearEnabledPrimaryMode() throws Exception { |
| atomicPut(true, new NearCacheConfiguration()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAtomicPutPrimaryMode() throws Exception { |
| atomicPut(false, null); |
| } |
| |
| /** |
| * @param putAll If {@code true} executes putAll. |
| * @param nearCfg Near cache configuration. |
| * @throws Exception If failed. |
| */ |
| private void atomicPut(final boolean putAll, |
| @Nullable NearCacheConfiguration nearCfg) throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(ATOMIC); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| |
| ccfg.setNearConfiguration(nearCfg); |
| |
| ccfg.setNearConfiguration(null); |
| |
| Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| final int KEYS = putAll ? 100 : 1; |
| |
| for (int i = 0; i < KEYS; i++) |
| map.put(i, i); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi(); |
| |
| // Block messages requests for both nodes. |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite1.localNode().id()); |
| |
| final IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| if (putAll) |
| cache.putAll(map); |
| else |
| cache.put(0, 0); |
| |
| return null; |
| } |
| }); |
| |
| assertFalse(putFut.isDone()); |
| |
| IgniteEx ignite3 = startGrid(3); |
| |
| awaitPartitionMapExchange(); |
| |
| log.info("Stop block1."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| checkData(map, null, cache, 4); |
| |
| ignite3.close(); |
| |
| map.clear(); |
| |
| for (int i = 0; i < KEYS; i++) |
| map.put(i, i + 1); |
| |
| // Block messages requests for single node. |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite0.localNode().id()); |
| |
| putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| if (putAll) |
| cache.putAll(map); |
| else |
| cache.put(0, 1); |
| |
| return null; |
| } |
| }); |
| |
| assertFalse(putFut.isDone()); |
| |
| startGrid(3); |
| |
| log.info("Stop block2."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| checkData(map, null, cache, 4); |
| |
| for (int i = 0; i < KEYS; i++) |
| map.put(i, i + 2); |
| |
| if (putAll) |
| cache.putAll(map); |
| else |
| cache.put(0, 2); |
| |
| checkData(map, null, cache, 4); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAtomicNoRemapPrimaryMode() throws Exception { |
| atomicNoRemap(); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| private void atomicNoRemap() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(ATOMIC); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| IgniteEx ignite2 = startGrid(2); |
| |
| Ignite ignite3 = startClientGrid(3); |
| |
| awaitPartitionMapExchange(); |
| |
| assertTrue(ignite3.configuration().isClientMode()); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| map.put(primaryKey(ignite0.cache(DEFAULT_CACHE_NAME)), 0); |
| map.put(primaryKey(ignite1.cache(DEFAULT_CACHE_NAME)), 1); |
| map.put(primaryKey(ignite2.cache(DEFAULT_CACHE_NAME)), 2); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi(); |
| |
| // Block messages requests for both nodes. |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite1.localNode().id()); |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite2.localNode().id()); |
| |
| spi.record(GridNearAtomicFullUpdateRequest.class); |
| |
| final IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| cache.putAll(map); |
| |
| return null; |
| } |
| }); |
| |
| IgniteEx ignite4 = startClientGrid(4); |
| |
| assertTrue(ignite4.configuration().isClientMode()); |
| |
| assertFalse(putFut.isDone()); |
| |
| log.info("Stop block."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| spi.record(null); |
| |
| checkData(map, null, cache, 5); |
| |
| List<Object> msgs = spi.recordedMessages(); |
| |
| assertEquals(3, msgs.size()); |
| |
| map.put(primaryKey(ignite0.cache(DEFAULT_CACHE_NAME)), 3); |
| map.put(primaryKey(ignite1.cache(DEFAULT_CACHE_NAME)), 4); |
| map.put(primaryKey(ignite2.cache(DEFAULT_CACHE_NAME)), 5); |
| |
| cache.putAll(map); |
| |
| checkData(map, null, cache, 5); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAtomicGetAndPutPrimaryMode() throws Exception { |
| atomicGetAndPut(); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| private void atomicGetAndPut() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(ATOMIC); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| |
| ignite0.cache(DEFAULT_CACHE_NAME).put(0, 0); |
| |
| Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| map.put(0, 1); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi(); |
| |
| // Block messages requests for both nodes. |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearAtomicFullUpdateRequest.class, ignite1.localNode().id()); |
| |
| final IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<Integer> putFut = GridTestUtils.runAsync(new Callable<Integer>() { |
| @Override public Integer call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| return cache.getAndPut(0, 1); |
| } |
| }); |
| |
| assertFalse(putFut.isDone()); |
| |
| startGrid(3); |
| |
| log.info("Stop block."); |
| |
| spi.stopBlock(); |
| |
| Integer old = putFut.get(); |
| |
| checkData(map, null, cache, 4); |
| |
| assertEquals((Object)0, old); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testTxPutAll() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| |
| Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi(); |
| |
| spi.blockMessages(GridNearTxPrepareRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearTxPrepareRequest.class, ignite1.localNode().id()); |
| |
| final IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| cache.putAll(map); |
| |
| return null; |
| } |
| }); |
| |
| assertFalse(putFut.isDone()); |
| |
| startGrid(3); |
| |
| log.info("Stop block."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| checkData(map, null, cache, 4); |
| |
| map.clear(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i + 1); |
| |
| cache.putAll(map); |
| |
| checkData(map, null, cache, 4); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTx() throws Exception { |
| pessimisticTx(null); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTxNearEnabled() throws Exception { |
| pessimisticTx(new NearCacheConfiguration()); |
| } |
| |
| /** |
| * @param nearCfg Near cache configuration. |
| * @throws Exception If failed. |
| */ |
| private void pessimisticTx(NearCacheConfiguration nearCfg) throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| ccfg.setNearConfiguration(nearCfg); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| |
| awaitPartitionMapExchange(); |
| |
| final Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi(); |
| |
| spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id()); |
| |
| spi.record(GridNearLockRequest.class); |
| |
| final IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| assertFalse(putFut.isDone()); |
| |
| IgniteEx ignite3 = startGrid(3); |
| |
| awaitPartitionMapExchange(); |
| |
| log.info("Stop block1."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| spi.record(null); |
| |
| checkData(map, null, cache, 4); |
| |
| List<Object> msgs = spi.recordedMessages(); |
| |
| assertTrue(((GridNearLockRequest)msgs.get(0)).firstClientRequest()); |
| assertTrue(((GridNearLockRequest)msgs.get(1)).firstClientRequest()); |
| |
| for (int i = 2; i < msgs.size(); i++) |
| assertFalse(((GridNearLockRequest)msgs.get(i)).firstClientRequest()); |
| |
| ignite3.close(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i + 1); |
| |
| spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id()); |
| |
| putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| for (Map.Entry<Integer, Integer> e : map.entrySet()) |
| cache.put(e.getKey(), e.getValue()); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| startGrid(3); |
| |
| log.info("Stop block2."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| awaitPartitionMapExchange(); |
| |
| checkData(map, null, cache, 4); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i + 2); |
| |
| try (Transaction tx = ignite2.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| checkData(map, null, cache, 4); |
| } |
| |
| /** |
| * Tries to find keys for two partitions: for one partition assignment should not change after node join, |
| * for another primary node should change. |
| * |
| * @param ignite Ignite. |
| * @param nodes Current nodes. |
| * @return Found keys. |
| */ |
| private IgniteBiTuple<Integer, Integer> findKeys(Ignite ignite, ClusterNode...nodes) { |
| ClusterNode newNode = new TcpDiscoveryNode(); |
| |
| GridTestUtils.setFieldValue(newNode, "consistentId", getTestIgniteInstanceName(4)); |
| GridTestUtils.setFieldValue(newNode, "id", UUID.randomUUID()); |
| |
| List<ClusterNode> topNodes = new ArrayList<>(); |
| |
| Collections.addAll(topNodes, nodes); |
| |
| topNodes.add(newNode); |
| |
| DiscoveryEvent discoEvt = new DiscoveryEvent(newNode, "", EventType.EVT_NODE_JOINED, newNode); |
| |
| final long topVer = ignite.cluster().topologyVersion(); |
| |
| GridAffinityFunctionContextImpl ctx = new GridAffinityFunctionContextImpl(topNodes, |
| null, |
| discoEvt, |
| new AffinityTopologyVersion(topVer + 1), |
| 1); |
| |
| AffinityFunction affFunc = ignite.cache(DEFAULT_CACHE_NAME).getConfiguration(CacheConfiguration.class).getAffinity(); |
| |
| List<List<ClusterNode>> newAff = affFunc.assignPartitions(ctx); |
| |
| List<List<ClusterNode>> curAff = ((IgniteKernal)ignite).context().cache().internalCache(DEFAULT_CACHE_NAME).context(). |
| affinity().assignments(new AffinityTopologyVersion(topVer)); |
| |
| Integer key1 = null; |
| Integer key2 = null; |
| |
| Affinity<Integer> aff = ignite.affinity(DEFAULT_CACHE_NAME); |
| |
| for (int i = 0; i < curAff.size(); i++) { |
| if (key1 == null) { |
| List<ClusterNode> oldNodes = curAff.get(i); |
| List<ClusterNode> newNodes = newAff.get(i); |
| |
| if (oldNodes.equals(newNodes)) |
| key1 = findKey(aff, i); |
| } |
| |
| if (key2 == null) { |
| ClusterNode oldPrimary = F.first(curAff.get(i)); |
| ClusterNode newPrimary = F.first(newAff.get(i)); |
| |
| if (!oldPrimary.equals(newPrimary)) |
| key2 = findKey(aff, i); |
| } |
| |
| if (key1 != null && key2 != null) |
| break; |
| } |
| |
| if (key1 == null || key2 == null) |
| fail("Failed to find nodes required for test."); |
| |
| return new IgniteBiTuple<>(key1, key2); |
| } |
| |
| /** |
| * @param aff Affinity. |
| * @param part Required key partition. |
| * @return Key. |
| */ |
| private Integer findKey(Affinity<Integer> aff, int part) { |
| for (int i = 0; i < 10_000; i++) { |
| Integer key = i; |
| |
| if (aff.partition(key) == part) |
| return key; |
| } |
| |
| fail(); |
| |
| return null; |
| } |
| |
| /** |
| * Tests specific scenario when mapping for first locked keys does not change, but changes for second one. |
| * |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTx2() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| IgniteEx ignite2 = startGrid(2); |
| |
| awaitPartitionMapExchange(); |
| |
| final Ignite ignite3 = startClientGrid(3); |
| |
| assertTrue(ignite3.configuration().isClientMode()); |
| |
| AffinityTopologyVersion topVer1 = new AffinityTopologyVersion(4, 0); |
| |
| assertEquals(topVer1, |
| ignite0.context().cache().internalCache(DEFAULT_CACHE_NAME).context().topology().readyTopologyVersion()); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi(); |
| |
| IgniteBiTuple<Integer, Integer> keys = |
| findKeys(ignite0, ignite0.localNode(), ignite1.localNode(), ignite2.localNode()); |
| |
| final Integer key1 = keys.get1(); |
| final Integer key2 = keys.get2(); |
| |
| spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite2.localNode().id()); |
| |
| final IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache.put(key1, 1); |
| cache.put(key2, 2); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| IgniteEx ignite4 = startGrid(4); |
| |
| int minorVer = ignite4.configuration().isLateAffinityAssignment() ? 1 : 0; |
| |
| AffinityTopologyVersion topVer2 = new AffinityTopologyVersion(5, minorVer); |
| |
| ignite0.context().cache().context().exchange().affinityReadyFuture(topVer2).get(); |
| |
| assertEquals(topVer2, ignite0.context().cache().internalCache(DEFAULT_CACHE_NAME).context().topology().readyTopologyVersion()); |
| |
| GridCacheAffinityManager aff = ignite0.context().cache().internalCache(DEFAULT_CACHE_NAME).context().affinity(); |
| |
| List<ClusterNode> nodes1 = aff.nodesByKey(key1, topVer1); |
| List<ClusterNode> nodes2 = aff.nodesByKey(key1, topVer2); |
| |
| assertEquals(nodes1, nodes2); |
| |
| nodes1 = aff.nodesByKey(key2, topVer1); |
| nodes2 = aff.nodesByKey(key2, topVer2); |
| |
| assertFalse(nodes1.get(0).equals(nodes2.get(0))); |
| |
| assertFalse(putFut.isDone()); |
| |
| log.info("Stop block."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| checkData(F.asMap(key1, 1, key2, 2), null, cache, 5); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTxNearEnabledNoRemap() throws Exception { |
| pessimisticTxNoRemap(new NearCacheConfiguration()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTxNoRemap() throws Exception { |
| pessimisticTxNoRemap(null); |
| } |
| |
| /** |
| * @param nearCfg Near cache configuration. |
| * @throws Exception If failed. |
| */ |
| private void pessimisticTxNoRemap(@Nullable NearCacheConfiguration nearCfg) throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| ccfg.setNearConfiguration(nearCfg); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| IgniteEx ignite2 = startGrid(2); |
| |
| final Ignite ignite3 = startClientGrid(3); |
| |
| assertTrue(ignite3.configuration().isClientMode()); |
| |
| awaitPartitionMapExchange(); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi(); |
| |
| for (int i = 0; i < 100; i++) |
| primaryCache(i, DEFAULT_CACHE_NAME).put(i, -1); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i); |
| |
| spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite2.localNode().id()); |
| |
| spi.record(GridNearLockRequest.class); |
| |
| final IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| for (Map.Entry<Integer, Integer> e : map.entrySet()) |
| cache.put(e.getKey(), e.getValue()); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| IgniteEx ignite4 = startClientGrid(4); |
| |
| assertTrue(ignite4.configuration().isClientMode()); |
| |
| assertFalse(putFut.isDone()); |
| |
| log.info("Stop block."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| spi.record(null); |
| |
| checkData(map, null, cache, 5); |
| |
| List<Object> msgs = spi.recordedMessages(); |
| |
| checkClientLockMessages(msgs, map.size()); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i + 1); |
| |
| try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| checkData(map, null, cache, 5); |
| } |
| |
| /** |
| * @return Cache configuration. |
| */ |
| private CacheConfiguration testPessimisticTx3Cfg() { |
| CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(0); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| ccfg.setAffinity(new RendezvousAffinityFunction(false, 16)); |
| |
| return ccfg; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTx3() throws Exception { |
| for (int iter = 0; iter < 5; iter++) { |
| info("Iteration: " + iter); |
| |
| ccfg = testPessimisticTx3Cfg(); |
| |
| IgniteEx ignite0 = startGrid(0); |
| |
| Map<Integer, Integer> map = new HashMap<>(); |
| |
| final IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME); |
| |
| for (int i = 0; i < 10000; i++) { |
| cache0.put(i, i); |
| map.put(i, i + 1); |
| } |
| |
| ccfg = testPessimisticTx3Cfg(); |
| |
| final Ignite ignite3 = startClientGrid(3); |
| |
| final IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi(); |
| spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id()); |
| |
| IgniteInternalFuture putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| spi.waitForBlocked(); |
| |
| ccfg = testPessimisticTx3Cfg(); |
| |
| startGrid(1); |
| |
| // Want provoke case when client req is processed when target partition is RENTING, |
| // there is no easy way to do it, so just try sleep. |
| U.sleep(ThreadLocalRandom.current().nextInt(1000) + 100); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| stopAllGrids(); |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testOptimisticSerializableTx() throws Exception { |
| optimisticSerializableTx(null); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testOptimisticSerializableTxNearEnabled() throws Exception { |
| optimisticSerializableTx(new NearCacheConfiguration()); |
| } |
| |
| /** |
| * @param nearCfg Near cache configuration. |
| * @throws Exception If failed. |
| */ |
| private void optimisticSerializableTx(NearCacheConfiguration nearCfg) throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| ccfg.setNearConfiguration(nearCfg); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| |
| awaitPartitionMapExchange(); |
| |
| final Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| final Map<Integer, Integer> map = new HashMap<>(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi(); |
| |
| spi.blockMessages(GridNearTxPrepareRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearTxPrepareRequest.class, ignite1.localNode().id()); |
| |
| spi.record(GridNearTxPrepareRequest.class); |
| |
| final IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite2.transactions().txStart(OPTIMISTIC, SERIALIZABLE)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| assertFalse(putFut.isDone()); |
| |
| IgniteEx ignite3 = startGrid(3); |
| |
| awaitPartitionMapExchange(); |
| |
| log.info("Stop block1."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| spi.record(null); |
| |
| checkData(map, null, cache, 4); |
| |
| List<Object> msgs = spi.recordedMessages(); |
| |
| for (Object msg : msgs) |
| assertTrue(((GridNearTxPrepareRequest)msg).firstClientRequest()); |
| |
| assertEquals(5, msgs.size()); |
| |
| ignite3.close(); |
| |
| awaitPartitionMapExchange(); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i + 1); |
| |
| spi.blockMessages(GridNearTxPrepareRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearTxPrepareRequest.class, ignite1.localNode().id()); |
| |
| spi.record(GridNearTxPrepareRequest.class); |
| |
| putFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| try (Transaction tx = ignite2.transactions().txStart(OPTIMISTIC, SERIALIZABLE)) { |
| for (Map.Entry<Integer, Integer> e : map.entrySet()) |
| cache.put(e.getKey(), e.getValue()); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }); |
| |
| startGrid(3); |
| |
| awaitPartitionMapExchange(); |
| |
| log.info("Stop block2."); |
| |
| spi.stopBlock(); |
| |
| putFut.get(); |
| |
| spi.record(null); |
| |
| msgs = spi.recordedMessages(); |
| |
| for (Object msg : msgs) |
| assertTrue(((GridNearTxPrepareRequest)msg).firstClientRequest()); |
| |
| assertEquals(5, msgs.size()); |
| |
| checkData(map, null, cache, 4); |
| |
| for (int i = 0; i < 100; i++) |
| map.put(i, i + 2); |
| |
| try (Transaction tx = ignite2.transactions().txStart(OPTIMISTIC, SERIALIZABLE)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| checkData(map, null, cache, 4); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLock() throws Exception { |
| lock(null); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLockNearEnabled() throws Exception { |
| lock(new NearCacheConfiguration()); |
| } |
| |
| /** |
| * @param nearCfg Near cache configuration. |
| * @throws Exception If failed. |
| */ |
| private void lock(NearCacheConfiguration nearCfg) throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| ccfg.setNearConfiguration(nearCfg); |
| |
| final IgniteEx ignite0 = startGrid(0); |
| final IgniteEx ignite1 = startGrid(1); |
| |
| awaitPartitionMapExchange(); |
| |
| final Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| final List<Integer> keys = new ArrayList<>(); |
| |
| for (int i = 0; i < 100; i++) |
| keys.add(i); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite2.configuration().getCommunicationSpi(); |
| |
| spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id()); |
| spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id()); |
| |
| final IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| final CountDownLatch lockedLatch = new CountDownLatch(1); |
| |
| final CountDownLatch unlockLatch = new CountDownLatch(1); |
| |
| IgniteInternalFuture<Lock> lockFut = GridTestUtils.runAsync(new Callable<Lock>() { |
| @Override public Lock call() throws Exception { |
| Thread.currentThread().setName("put-thread"); |
| |
| Lock lock = cache.lockAll(keys); |
| |
| lock.lock(); |
| |
| log.info("Locked"); |
| |
| lockedLatch.countDown(); |
| |
| unlockLatch.await(); |
| |
| lock.unlock(); |
| |
| return lock; |
| } |
| }); |
| |
| startGrid(3); |
| |
| awaitPartitionMapExchange(); |
| |
| log.info("Stop block."); |
| |
| assertEquals(1, lockedLatch.getCount()); |
| |
| spi.stopBlock(); |
| |
| assertTrue(lockedLatch.await(3000, TimeUnit.MILLISECONDS)); |
| |
| IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME); |
| |
| for (Integer key : keys) { |
| Lock lock = cache0.lock(key); |
| |
| assertFalse(lock.tryLock()); |
| } |
| |
| unlockLatch.countDown(); |
| |
| lockFut.get(); |
| |
| awaitPartitionMapExchange(); |
| |
| boolean wait = GridTestUtils.waitForCondition(new GridAbsPredicate() { |
| @Override public boolean apply() { |
| for (int i = 0; i < 4; i++) { |
| if (!unlocked(ignite(i))) |
| return false; |
| } |
| |
| return true; |
| } |
| |
| private boolean unlocked(Ignite ignite) { |
| IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); |
| |
| for (Integer key : keys) { |
| if (cache.isLocalLocked(key, false)) { |
| log.info("Key is locked [key=" + key + ", node=" + ignite.name() + ']'); |
| |
| return false; |
| } |
| } |
| |
| return true; |
| } |
| }, 10_000); |
| |
| assertTrue(wait); |
| |
| for (Integer key : keys) { |
| Lock lock = cache0.lock(key); |
| |
| assertTrue("Failed to lock: " + key, lock.tryLock()); |
| |
| lock.unlock(); |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTxMessageClientFirstFlag() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| IgniteEx ignite2 = startGrid(2); |
| |
| awaitPartitionMapExchange(); |
| |
| Ignite ignite3 = startClientGrid(3); |
| |
| assertTrue(ignite3.configuration().isClientMode()); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi(); |
| |
| spi.record(GridNearLockRequest.class); |
| |
| IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME); |
| |
| Affinity<Integer> aff = ignite0.affinity(DEFAULT_CACHE_NAME); |
| |
| try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| Integer key1 = findKey(aff, 1); |
| Integer key2 = findKey(aff, 2); |
| Integer key3 = findKey(aff, 3); |
| |
| cache.put(key1, 1); |
| cache.put(key2, 2); |
| cache.put(key3, 3); |
| |
| tx.commit(); |
| } |
| |
| checkClientLockMessages(spi.recordedMessages(), 3); |
| |
| Map<Integer, Integer> map = new LinkedHashMap<>(); |
| |
| map.put(primaryKey(ignite0.cache(DEFAULT_CACHE_NAME)), 4); |
| map.put(primaryKey(ignite1.cache(DEFAULT_CACHE_NAME)), 5); |
| map.put(primaryKey(ignite2.cache(DEFAULT_CACHE_NAME)), 6); |
| map.put(primaryKeys(ignite0.cache(DEFAULT_CACHE_NAME), 1, 10_000).get(0), 7); |
| |
| try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| |
| checkClientLockMessages(spi.recordedMessages(), 4); |
| |
| spi.record(null); |
| |
| TestCommunicationSpi spi0 = (TestCommunicationSpi)ignite0.configuration().getCommunicationSpi(); |
| |
| spi0.record(GridNearLockRequest.class); |
| |
| List<Integer> keys = primaryKeys(ignite1.cache(DEFAULT_CACHE_NAME), 3, 0); |
| |
| IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME); |
| |
| try (Transaction tx = ignite0.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) { |
| cache0.put(keys.get(0), 0); |
| cache0.put(keys.get(1), 1); |
| cache0.put(keys.get(2), 2); |
| |
| tx.commit(); |
| } |
| |
| List<Object> msgs = spi0.recordedMessages(); |
| |
| assertEquals(3, msgs.size()); |
| |
| for (Object msg : msgs) |
| assertFalse(((GridNearLockRequest)msg).firstClientRequest()); |
| } |
| |
| /** |
| * @param msgs Messages. |
| * @param expCnt Expected number of messages. |
| */ |
| private void checkClientLockMessages(List<Object> msgs, int expCnt) { |
| assertEquals(expCnt, msgs.size()); |
| |
| assertTrue(((GridNearLockRequest)msgs.get(0)).firstClientRequest()); |
| |
| for (int i = 1; i < msgs.size(); i++) |
| assertFalse(((GridNearLockRequest)msgs.get(i)).firstClientRequest()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testOptimisticTxMessageClientFirstFlag() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| IgniteEx ignite2 = startGrid(2); |
| |
| awaitPartitionMapExchange(); |
| |
| Ignite ignite3 = startClientGrid(3); |
| |
| assertTrue(ignite3.configuration().isClientMode()); |
| |
| TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi(); |
| |
| IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME); |
| |
| List<Integer> keys0 = primaryKeys(ignite0.cache(DEFAULT_CACHE_NAME), 2, 0); |
| List<Integer> keys1 = primaryKeys(ignite1.cache(DEFAULT_CACHE_NAME), 2, 0); |
| List<Integer> keys2 = primaryKeys(ignite2.cache(DEFAULT_CACHE_NAME), 2, 0); |
| |
| LinkedHashMap<Integer, Integer> map = new LinkedHashMap<>(); |
| |
| map.put(keys0.get(0), 1); |
| map.put(keys1.get(0), 2); |
| map.put(keys2.get(0), 3); |
| map.put(keys0.get(1), 4); |
| map.put(keys1.get(1), 5); |
| map.put(keys2.get(1), 6); |
| |
| spi.record(GridNearTxPrepareRequest.class); |
| |
| try (Transaction tx = ignite3.transactions().txStart(OPTIMISTIC, REPEATABLE_READ)) { |
| for (Map.Entry<Integer, Integer> e : map.entrySet()) |
| cache.put(e.getKey(), e.getValue()); |
| |
| tx.commit(); |
| } |
| |
| checkClientPrepareMessages(spi.recordedMessages(), 6); |
| |
| checkData(map, null, cache, 4); |
| |
| cache.putAll(map); |
| |
| checkClientPrepareMessages(spi.recordedMessages(), 6); |
| |
| spi.record(null); |
| |
| checkData(map, null, cache, 4); |
| |
| IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME); |
| |
| TestCommunicationSpi spi0 = (TestCommunicationSpi)ignite0.configuration().getCommunicationSpi(); |
| |
| spi0.record(GridNearTxPrepareRequest.class); |
| |
| cache0.putAll(map); |
| |
| spi0.record(null); |
| |
| List<Object> msgs = spi0.recordedMessages(); |
| |
| assertEquals(4, msgs.size()); |
| |
| for (Object msg : msgs) |
| assertFalse(((GridNearTxPrepareRequest)msg).firstClientRequest()); |
| |
| checkData(map, null, cache, 4); |
| } |
| |
| /** |
| * @param msgs Messages. |
| * @param expCnt Expected number of messages. |
| */ |
| private void checkClientPrepareMessages(List<Object> msgs, int expCnt) { |
| assertEquals(expCnt, msgs.size()); |
| |
| assertTrue(((GridNearTxPrepareRequest)msgs.get(0)).firstClientRequest()); |
| |
| for (int i = 1; i < msgs.size(); i++) |
| assertFalse(((GridNearTxPrepareRequest) msgs.get(i)).firstClientRequest()); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLockRemoveAfterClientFailed() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| IgniteEx ignite0 = startGrid(0); |
| IgniteEx ignite1 = startGrid(1); |
| |
| Ignite ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| IgniteCache<Integer, Integer> cache2 = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| final Integer key = 0; |
| |
| Lock lock2 = cache2.lock(key); |
| |
| lock2.lock(); |
| |
| ignite2.close(); |
| |
| IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME); |
| |
| assertFalse(cache0.isLocalLocked(key, false)); |
| |
| IgniteCache<Integer, Integer> cache1 = ignite1.cache(DEFAULT_CACHE_NAME); |
| |
| assertFalse(cache1.isLocalLocked(key, false)); |
| |
| Lock lock1 = cache1.lock(0); |
| |
| assertTrue(lock1.tryLock(5000, TimeUnit.MILLISECONDS)); |
| |
| lock1.unlock(); |
| |
| ignite2 = startClientGrid(2); |
| |
| assertTrue(ignite2.configuration().isClientMode()); |
| |
| cache2 = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| lock2 = cache2.lock(0); |
| |
| assertTrue(lock2.tryLock(5000, TimeUnit.MILLISECONDS)); |
| |
| lock2.unlock(); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLockFromClientBlocksExchange() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(TRANSACTIONAL); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| startGrid(0); |
| startGrid(1); |
| |
| Ignite ignite2 = startClientGrid(2); |
| |
| IgniteCache<Integer, Integer> cache = ignite2.cache(DEFAULT_CACHE_NAME); |
| |
| Lock lock = cache.lock(0); |
| |
| lock.lock(); |
| |
| IgniteInternalFuture<?> startFut = GridTestUtils.runAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| startGrid(3); |
| |
| return null; |
| } |
| }); |
| |
| U.sleep(2000); |
| |
| assertFalse(startFut.isDone()); |
| |
| AffinityTopologyVersion ver = new AffinityTopologyVersion(4); |
| |
| List<IgniteInternalFuture<?>> futs = new ArrayList<>(); |
| |
| U.sleep(2000); |
| |
| for (int i = 0; i < 3; i++) { |
| Ignite ignite = ignite(i); |
| |
| IgniteInternalFuture<?> fut = |
| ((IgniteKernal)ignite).context().cache().context().exchange().affinityReadyFuture(ver); |
| |
| assertNotNull(fut); |
| |
| assertFalse(fut.isDone()); |
| |
| futs.add(fut); |
| } |
| |
| lock.unlock(); |
| |
| for (IgniteInternalFuture<?> fut : futs) |
| fut.get(10_000); |
| |
| startFut.get(10_000); |
| } |
| |
| /** |
| * @param map Expected data. |
| * @param keys Expected keys (if expected data is not specified). |
| * @param clientCache Client cache. |
| * @param expNodes Expected nodes number. |
| * @throws Exception If failed. |
| */ |
| private void checkData(final Map<Integer, Integer> map, |
| final Set<Integer> keys, |
| IgniteCache<?, ?> clientCache, |
| final int expNodes) |
| throws Exception |
| { |
| final List<Ignite> nodes = G.allGrids(); |
| |
| final Affinity<Integer> aff = nodes.get(0).affinity(DEFAULT_CACHE_NAME); |
| |
| assertEquals(expNodes, nodes.size()); |
| |
| boolean hasNearCache = clientCache.getConfiguration(CacheConfiguration.class).getNearConfiguration() != null; |
| |
| final Ignite nearCacheNode = hasNearCache ? clientCache.unwrap(Ignite.class) : null; |
| |
| boolean wait = GridTestUtils.waitForCondition(new PA() { |
| @Override public boolean apply() { |
| try { |
| Set<Integer> keys0 = map != null ? map.keySet() : keys; |
| |
| assertNotNull(keys0); |
| |
| for (Integer key : keys0) { |
| GridCacheVersion ver = null; |
| Object val = null; |
| |
| for (Ignite node : nodes) { |
| IgniteCache<Integer, Integer> cache = node.cache(DEFAULT_CACHE_NAME); |
| |
| boolean affNode = aff.isPrimaryOrBackup(node.cluster().localNode(), key); |
| |
| Object val0 = cache.localPeek(key); |
| |
| if (affNode || node == nearCacheNode) { |
| if (map != null) |
| assertEquals("Unexpected value for " + node.name(), map.get(key), val0); |
| else |
| assertNotNull("Unexpected value for " + node.name(), val0); |
| |
| GridCacheAdapter cache0 = ((IgniteKernal)node).internalCache(DEFAULT_CACHE_NAME); |
| |
| if (affNode && cache0.isNear()) |
| cache0 = ((GridNearCacheAdapter)cache0).dht(); |
| |
| GridCacheEntryEx entry = cache0.entryEx(key); |
| |
| try { |
| entry.unswap(true); |
| |
| assertNotNull("No entry [node=" + node.name() + ", key=" + key + ']', entry); |
| |
| GridCacheVersion ver0 = entry instanceof GridNearCacheEntry ? |
| ((GridNearCacheEntry)entry).dhtVersion() : entry.version(); |
| |
| assertNotNull("Null version [node=" + node.name() + ", key=" + key + ']', ver0); |
| |
| if (ver == null) { |
| ver = ver0; |
| val = val0; |
| } |
| else { |
| assertEquals("Version check failed [node=" + node.name() + |
| ", key=" + key + |
| ", affNode=" + affNode + |
| ", primary=" + aff.isPrimary(node.cluster().localNode(), key) + ']', |
| ver0, |
| ver); |
| |
| assertEquals("Value check failed [node=" + node.name() + |
| ", key=" + key + |
| ", affNode=" + affNode + |
| ", primary=" + aff.isPrimary(node.cluster().localNode(), key) + ']', |
| val0, |
| val); |
| } |
| } |
| finally { |
| entry.touch(); |
| } |
| } |
| else |
| assertNull("Unexpected non-null value for " + node.name(), val0); |
| } |
| } |
| } |
| catch (AssertionError e) { |
| log.info("Check failed, will retry: " + e); |
| |
| return false; |
| } |
| catch (Exception e) { |
| fail("Unexpected exception: " + e); |
| } |
| |
| return true; |
| } |
| }, 10_000); |
| |
| assertTrue("Data check failed.", wait); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testAtomicPrimaryPutAllMultinode() throws Exception { |
| multinode(ATOMIC, TestType.PUT_ALL); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testOptimisticTxPutAllMultinode() throws Exception { |
| multinode(TRANSACTIONAL, TestType.OPTIMISTIC_TX); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testOptimisticSerializableTxPutAllMultinode() throws Exception { |
| multinode(TRANSACTIONAL, TestType.OPTIMISTIC_SERIALIZABLE_TX); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testPessimisticTxPutAllMultinode() throws Exception { |
| multinode(TRANSACTIONAL, TestType.PESSIMISTIC_TX); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLockAllMultinode() throws Exception { |
| multinode(TRANSACTIONAL, TestType.LOCK); |
| } |
| |
| /** |
| * @param atomicityMode Atomicity mode cache. |
| * @param testType Test type. |
| * @throws Exception If failed. |
| */ |
| private void multinode(CacheAtomicityMode atomicityMode, final TestType testType) |
| throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(atomicityMode); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| final int SRV_CNT = 4; |
| |
| for (int i = 0; i < SRV_CNT; i++) |
| startGrid(i); |
| |
| final int CLIENT_CNT = 4; |
| |
| final List<Ignite> clients = new ArrayList<>(); |
| |
| for (int i = 0; i < CLIENT_CNT; i++) { |
| Ignite ignite = startClientGrid(SRV_CNT + i); |
| |
| assertTrue(ignite.configuration().isClientMode()); |
| |
| clients.add(ignite); |
| } |
| |
| final AtomicBoolean stop = new AtomicBoolean(); |
| |
| final AtomicInteger threadIdx = new AtomicInteger(0); |
| |
| final int THREADS = CLIENT_CNT * 3; |
| |
| final GridConcurrentHashSet<Integer> putKeys = new GridConcurrentHashSet<>(); |
| |
| IgniteInternalFuture<?> fut; |
| |
| try { |
| fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| int clientIdx = threadIdx.getAndIncrement() % CLIENT_CNT; |
| |
| Ignite ignite = clients.get(clientIdx); |
| |
| assertTrue(ignite.configuration().isClientMode()); |
| |
| Thread.currentThread().setName("update-thread-" + ignite.name()); |
| |
| IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); |
| |
| boolean useTx = testType == TestType.OPTIMISTIC_TX || |
| testType == TestType.OPTIMISTIC_SERIALIZABLE_TX || |
| testType == TestType.PESSIMISTIC_TX; |
| |
| if (useTx || testType == TestType.LOCK) { |
| assertEquals(TRANSACTIONAL, |
| cache.getConfiguration(CacheConfiguration.class).getAtomicityMode()); |
| } |
| |
| ThreadLocalRandom rnd = ThreadLocalRandom.current(); |
| |
| int cntr = 0; |
| |
| while (!stop.get()) { |
| TreeMap<Integer, Integer> map = new TreeMap<>(); |
| |
| for (int i = 0; i < 100; i++) { |
| Integer key = rnd.nextInt(0, 1000); |
| |
| map.put(key, rnd.nextInt()); |
| } |
| |
| try { |
| if (testType == TestType.LOCK) { |
| Lock lock = cache.lockAll(map.keySet()); |
| |
| lock.lock(); |
| |
| lock.unlock(); |
| } |
| else { |
| if (useTx) { |
| IgniteTransactions txs = ignite.transactions(); |
| |
| TransactionConcurrency concurrency = |
| testType == TestType.PESSIMISTIC_TX ? PESSIMISTIC : OPTIMISTIC; |
| |
| TransactionIsolation isolation = testType == TestType.OPTIMISTIC_SERIALIZABLE_TX ? |
| SERIALIZABLE : REPEATABLE_READ; |
| |
| try (Transaction tx = txs.txStart(concurrency, isolation)) { |
| cache.putAll(map); |
| |
| tx.commit(); |
| } |
| } |
| else |
| cache.putAll(map); |
| |
| putKeys.addAll(map.keySet()); |
| } |
| } |
| catch (CacheException | IgniteException e) { |
| log.info("Operation failed, ignore: " + e); |
| } |
| |
| if (++cntr % 100 == 0) |
| log.info("Iteration: " + cntr); |
| |
| if (updateBarrier != null) |
| updateBarrier.await(); |
| } |
| |
| return null; |
| } |
| }, THREADS, "update-thread"); |
| |
| long stopTime = System.currentTimeMillis() + 60_000; |
| |
| while (System.currentTimeMillis() < stopTime) { |
| boolean restartClient = ThreadLocalRandom.current().nextBoolean(); |
| |
| Integer idx = null; |
| |
| if (restartClient) { |
| log.info("Start client node."); |
| |
| IgniteEx ignite = startClientGrid(SRV_CNT + CLIENT_CNT); |
| |
| IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); |
| |
| assertNotNull(cache); |
| } |
| else { |
| idx = ThreadLocalRandom.current().nextInt(0, SRV_CNT); |
| |
| log.info("Stop server node: " + idx); |
| |
| stopGrid(idx); |
| } |
| |
| updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() { |
| @Override public void run() { |
| updateBarrier = null; |
| } |
| }); |
| |
| try { |
| updateBarrier.await(30_000, TimeUnit.MILLISECONDS); |
| } |
| catch (TimeoutException ignored) { |
| log.error("Failed to wait for update."); |
| |
| for (Ignite ignite : G.allGrids()) |
| ((IgniteKernal)ignite).dumpDebugInfo(); |
| |
| U.dumpThreads(log); |
| |
| CyclicBarrier barrier0 = updateBarrier; |
| |
| if (barrier0 != null) |
| barrier0.reset(); |
| |
| fail("Failed to wait for update."); |
| } |
| |
| U.sleep(500); |
| |
| if (restartClient) { |
| log.info("Stop client node."); |
| |
| stopGrid(SRV_CNT + CLIENT_CNT); |
| } |
| else { |
| log.info("Start server node: " + idx); |
| |
| startGrid(idx); |
| } |
| |
| updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() { |
| @Override public void run() { |
| updateBarrier = null; |
| } |
| }); |
| |
| try { |
| updateBarrier.await(30_000, TimeUnit.MILLISECONDS); |
| } |
| catch (TimeoutException ignored) { |
| log.error("Failed to wait for update."); |
| |
| for (Ignite ignite : G.allGrids()) |
| ((IgniteKernal)ignite).dumpDebugInfo(); |
| |
| U.dumpThreads(log); |
| |
| CyclicBarrier barrier0 = updateBarrier; |
| |
| if (barrier0 != null) |
| barrier0.reset(); |
| |
| fail("Failed to wait for update."); |
| } |
| |
| U.sleep(500); |
| } |
| } |
| finally { |
| stop.set(true); |
| } |
| |
| fut.get(30_000); |
| |
| if (testType != TestType.LOCK) |
| checkData(null, putKeys, grid(SRV_CNT).cache(DEFAULT_CACHE_NAME), SRV_CNT + CLIENT_CNT); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testServersLeaveOnStart() throws Exception { |
| ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); |
| |
| ccfg.setCacheMode(PARTITIONED); |
| ccfg.setBackups(1); |
| ccfg.setAtomicityMode(ATOMIC); |
| ccfg.setWriteSynchronizationMode(FULL_SYNC); |
| ccfg.setRebalanceMode(SYNC); |
| |
| Ignite ignite0 = startGrid(0); |
| |
| final AtomicInteger nodeIdx = new AtomicInteger(2); |
| |
| final int CLIENTS = 10; |
| |
| IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Void>() { |
| @Override public Void call() throws Exception { |
| int idx = nodeIdx.getAndIncrement(); |
| |
| startClientGrid(idx); |
| |
| return null; |
| } |
| }, CLIENTS, "start-client"); |
| |
| ignite0.close(); |
| |
| fut.get(); |
| |
| for (int i = 0; i < CLIENTS; i++) { |
| Ignite ignite = grid(i + 2); |
| |
| assertEquals(CLIENTS, ignite.cluster().nodes().size()); |
| } |
| |
| startGrid(0); |
| startGrid(1); |
| |
| awaitPartitionMapExchange(); |
| |
| for (int i = 0; i < CLIENTS; i++) { |
| Ignite ignite = grid(i + 2); |
| |
| IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); |
| |
| cache.put(i, i); |
| |
| assertEquals((Object)i, cache.get(i)); |
| } |
| } |
| |
| /** |
| * |
| */ |
| private static class TestCommunicationSpi extends TcpCommunicationSpi { |
| /** */ |
| @LoggerResource |
| private IgniteLogger log; |
| |
| /** */ |
| private List<T2<ClusterNode, GridIoMessage>> blockedMsgs = new ArrayList<>(); |
| |
| /** */ |
| private Map<Class<?>, Set<UUID>> blockCls = new HashMap<>(); |
| |
| /** */ |
| private Class<?> recordCls; |
| |
| /** */ |
| private List<Object> recordedMsgs = new ArrayList<>(); |
| |
| /** {@inheritDoc} */ |
| @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) |
| throws IgniteSpiException { |
| if (msg instanceof GridIoMessage) { |
| Object msg0 = ((GridIoMessage)msg).message(); |
| |
| synchronized (this) { |
| if (recordCls != null && msg0.getClass().equals(recordCls)) |
| recordedMsgs.add(msg0); |
| |
| Set<UUID> blockNodes = blockCls.get(msg0.getClass()); |
| |
| if (F.contains(blockNodes, node.id())) { |
| log.info("Block message [node=" + |
| node.attribute(IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME) + ", msg=" + msg0 + ']'); |
| |
| blockedMsgs.add(new T2<>(node, (GridIoMessage)msg)); |
| |
| notifyAll(); |
| |
| return; |
| } |
| } |
| } |
| |
| super.sendMessage(node, msg, ackC); |
| } |
| |
| /** |
| * @param recordCls Message class to record. |
| */ |
| void record(@Nullable Class<?> recordCls) { |
| synchronized (this) { |
| this.recordCls = recordCls; |
| } |
| } |
| |
| /** |
| * @return Recorded messages. |
| */ |
| List<Object> recordedMessages() { |
| synchronized (this) { |
| List<Object> msgs = recordedMsgs; |
| |
| recordedMsgs = new ArrayList<>(); |
| |
| return msgs; |
| } |
| } |
| |
| /** |
| * @param cls Message class. |
| * @param nodeId Node ID. |
| */ |
| void blockMessages(Class<?> cls, UUID nodeId) { |
| synchronized (this) { |
| Set<UUID> set = blockCls.get(cls); |
| |
| if (set == null) { |
| set = new HashSet<>(); |
| |
| blockCls.put(cls, set); |
| } |
| |
| set.add(nodeId); |
| } |
| } |
| |
| /** |
| * |
| */ |
| void stopBlock() { |
| synchronized (this) { |
| blockCls.clear(); |
| |
| for (T2<ClusterNode, GridIoMessage> msg : blockedMsgs) { |
| ClusterNode node = msg.get1(); |
| |
| log.info("Send blocked message: [node=" + |
| node.attribute(IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME) + |
| ", msg=" + msg.get2().message() + ']'); |
| |
| super.sendMessage(msg.get1(), msg.get2()); |
| } |
| |
| blockedMsgs.clear(); |
| } |
| } |
| |
| /** |
| * @throws InterruptedException If interrupted. |
| */ |
| public void waitForBlocked() throws InterruptedException { |
| synchronized (this) { |
| while (blockedMsgs.isEmpty()) |
| wait(); |
| } |
| } |
| } |
| |
| /** |
| * |
| */ |
| enum TestType { |
| /** */ |
| PUT_ALL, |
| |
| /** */ |
| OPTIMISTIC_TX, |
| |
| /** */ |
| OPTIMISTIC_SERIALIZABLE_TX, |
| |
| /** */ |
| PESSIMISTIC_TX, |
| |
| /** */ |
| LOCK |
| } |
| } |
