| /* |
| * 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.geode.disttx; |
| |
| import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.CountDownLatch; |
| |
| import org.junit.Ignore; |
| import org.junit.Test; |
| |
| import org.apache.geode.LogWriter; |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheTransactionManager; |
| import org.apache.geode.cache.CommitConflictException; |
| import org.apache.geode.cache.CommitIncompleteException; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.DiskStore; |
| import org.apache.geode.cache.InterestPolicy; |
| import org.apache.geode.cache.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionAttributes; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.SubscriptionAttributes; |
| import org.apache.geode.cache.server.CacheServer; |
| import org.apache.geode.internal.cache.BucketRegion; |
| import org.apache.geode.internal.cache.DistTXState; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.RegionEntry; |
| import org.apache.geode.internal.cache.TXManagerImpl; |
| import org.apache.geode.internal.cache.TXStateInterface; |
| import org.apache.geode.internal.cache.TXStateProxyImpl; |
| import org.apache.geode.internal.cache.execute.CustomerIDPartitionResolver; |
| import org.apache.geode.internal.cache.execute.data.CustId; |
| import org.apache.geode.internal.cache.execute.data.Customer; |
| import org.apache.geode.internal.cache.execute.data.Order; |
| import org.apache.geode.internal.cache.execute.data.OrderId; |
| import org.apache.geode.test.dunit.Assert; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.Invoke; |
| import org.apache.geode.test.dunit.LogWriterUtils; |
| import org.apache.geode.test.dunit.SerializableCallable; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase; |
| import org.apache.geode.util.internal.GeodeGlossary; |
| |
| @SuppressWarnings("deprecation") |
| |
| public class DistributedTransactionDUnitTest extends JUnit4CacheTestCase { |
| |
| protected final String CUSTOMER_PR = "customerPRRegion"; |
| protected final String ORDER_PR = "orderPRRegion"; |
| protected final String D_REFERENCE = "distrReference"; |
| protected final String PERSISTENT_CUSTOMER_PR = "persistentCustomerPRRegion"; |
| |
| protected final String CUSTOMER_RR = "customerRRRegion"; |
| |
| @Override |
| public final void postSetUp() throws Exception { |
| Invoke.invokeInEveryVM(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| System.setProperty(GeodeGlossary.GEMFIRE_PREFIX + "sync-commits", "true"); |
| return null; |
| } |
| }); |
| |
| Invoke.invokeInEveryVM(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| // System.setProperty("gemfire.ALLOW_PERSISTENT_TRANSACTIONS", "true"); |
| TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = true; |
| return null; |
| } |
| }); |
| } |
| |
| @Override |
| public final void preTearDownCacheTestCase() throws Exception { |
| Invoke.invokeInEveryVM(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| System.setProperty(GeodeGlossary.GEMFIRE_PREFIX + "sync-commits", "false"); |
| return null; |
| } |
| }); |
| Invoke.invokeInEveryVM(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| // System.setProperty("gemfire.ALLOW_PERSISTENT_TRANSACTIONS", "false"); |
| TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = false; |
| return null; |
| } |
| }); |
| } |
| |
| public DistributedTransactionDUnitTest() { |
| super(); |
| } |
| |
| public Object execute(VM vm, SerializableCallable c) { |
| return vm.invoke(c); |
| } |
| |
| public void execute(VM[] vms, SerializableCallable c) { |
| for (VM vm : vms) { |
| execute(vm, c); |
| } |
| } |
| |
| public int startServer(VM vm) { |
| return (Integer) vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| int port = getRandomAvailableTCPPort(); |
| CacheServer s = getCache().addCacheServer(); |
| s.setPort(port); |
| s.start(); |
| TXManagerImpl txMgr = (TXManagerImpl) getCache().getCacheTransactionManager(); |
| txMgr.setTransactionTimeToLiveForTest(10); |
| return port; |
| } |
| }); |
| } |
| |
| protected boolean getConcurrencyChecksEnabled() { |
| return true; |
| } |
| |
| void createRR(boolean isEmpty) { |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.DISTRIBUTED_ACK); |
| if (!isEmpty) { |
| af.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE); |
| } else { |
| af.setDataPolicy(DataPolicy.EMPTY); // for accessor |
| } |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| getCache().createRegion(CUSTOMER_RR, af.create()); |
| } |
| |
| void createPR(boolean accessor, int redundantCopies, InterestPolicy interestPolicy) { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| if (interestPolicy != null) { |
| af.setSubscriptionAttributes(new SubscriptionAttributes(interestPolicy)); |
| } |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(accessor ? 0 : 1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver1")) |
| .setRedundantCopies(redundantCopies).create()); |
| getCache().createRegion(CUSTOMER_PR, af.create()); |
| |
| } |
| |
| @Override |
| public Properties getDistributedSystemProperties() { |
| Properties props = super.getDistributedSystemProperties(); |
| return props; |
| } |
| |
| |
| void createRegions(boolean accessor, int redundantCopies, InterestPolicy interestPolicy) { |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.DISTRIBUTED_ACK); |
| af.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE); |
| // af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| getCache().createRegion(D_REFERENCE, af.create()); |
| |
| af = new AttributesFactory(); |
| // af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setDataPolicy(DataPolicy.PERSISTENT_PARTITION); |
| if (interestPolicy != null) { |
| af.setSubscriptionAttributes(new SubscriptionAttributes(interestPolicy)); |
| } |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(accessor ? 0 : 1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver1")) |
| .setRedundantCopies(redundantCopies).create()); |
| getCache().createRegion(CUSTOMER_PR, af.create()); |
| af.setDataPolicy(DataPolicy.PERSISTENT_PARTITION); |
| af.setPartitionAttributes(new PartitionAttributesFactory<OrderId, Order>().setTotalNumBuckets(4) |
| .setLocalMaxMemory(accessor ? 0 : 1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver2")) |
| .setRedundantCopies(redundantCopies).setColocatedWith(CUSTOMER_PR).create()); |
| getCache().createRegion(ORDER_PR, af.create()); |
| } |
| |
| public void createRegions(VM[] vms) { |
| for (VM vm : vms) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createRegions(false, 1, null); |
| return null; |
| } |
| |
| }); |
| } |
| } |
| |
| public void createRegions(final VM vm, final boolean accessor, final int redundantCopies) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createRegions(accessor, redundantCopies, null); |
| return null; |
| } |
| }); |
| } |
| |
| public void createRR(VM[] vms) { |
| for (VM vm : vms) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createRR(false); |
| return null; |
| } |
| }); |
| } |
| } |
| |
| public void createRRonAccessor(VM accessor) { |
| accessor.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createRR(true); |
| return null; |
| } |
| }); |
| } |
| |
| public void createPR(VM[] vms) { |
| for (VM vm : vms) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createPR(false, 0, null); |
| return null; |
| } |
| }); |
| } |
| } |
| |
| public void createPRwithRedundanyCopies(VM[] vms, final int redundency) { |
| for (VM vm : vms) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createPR(false, redundency, null); |
| return null; |
| } |
| }); |
| } |
| } |
| |
| public void createPRonAccessor(VM accessor, final int redundency) { |
| accessor.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createPR(true, redundency, null); |
| return null; |
| } |
| }); |
| } |
| |
| public void createPersistentPR(VM[] vms) { |
| for (VM vm : vms) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| createPersistentPR(); |
| return null; |
| } |
| }); |
| } |
| } |
| |
| public void createPersistentPR() { |
| getCache().createRegion(PERSISTENT_CUSTOMER_PR, |
| getPersistentPRAttributes(1, -1, getCache(), 113, true)); |
| } |
| |
| protected RegionAttributes getPersistentPRAttributes(final int redundancy, |
| final int recoveryDelay, Cache cache, int numBuckets, boolean synchronous) { |
| DiskStore ds = cache.findDiskStore("disk"); |
| if (ds == null) { |
| ds = cache.createDiskStoreFactory().setDiskDirs(getDiskDirs()).create("disk"); |
| } |
| AttributesFactory af = new AttributesFactory(); |
| PartitionAttributesFactory paf = new PartitionAttributesFactory(); |
| paf.setRedundantCopies(redundancy); |
| paf.setRecoveryDelay(recoveryDelay); |
| paf.setTotalNumBuckets(numBuckets); |
| paf.setLocalMaxMemory(500); |
| af.setPartitionAttributes(paf.create()); |
| af.setDataPolicy(DataPolicy.PERSISTENT_PARTITION); |
| af.setDiskStoreName("disk"); |
| af.setDiskSynchronous(synchronous); |
| RegionAttributes attr = af.create(); |
| return attr; |
| } |
| |
| |
| void populateData() { |
| Region custRegion = getCache().getRegion(CUSTOMER_PR); |
| Region orderRegion = getCache().getRegion(ORDER_PR); |
| Region refRegion = getCache().getRegion(D_REFERENCE); |
| for (int i = 0; i < 5; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custRegion.put(custId, customer); |
| orderRegion.put(orderId, order); |
| refRegion.put(custId, customer); |
| } |
| } |
| |
| void populateRR() { |
| Region custRegion = getCache().getRegion(CUSTOMER_RR); |
| for (int i = 0; i < 5; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| custRegion.put(custId, customer); |
| } |
| } |
| |
| public void populateData(final VM vm) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| populateData(); |
| return null; |
| } |
| }); |
| } |
| |
| public void populateRR(final VM vm) { |
| vm.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| populateRR(); |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testTransactionalPutOnReplicatedRegion() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createRR(new VM[] {server1, server2, server3}); |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| mgr.commit(); |
| mgr.begin(); |
| |
| Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER_RR); |
| CustId custId = new CustId(1); |
| Customer expectedCustomer = custRegion.get(custId); |
| assertNull(expectedCustomer); |
| |
| // Perform a put |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| custRegion.put(custIdOne, customerOne); |
| |
| // Rollback the transaction |
| mgr.rollback(); |
| |
| |
| mgr.begin(); |
| // Verify that the entry is rolled back |
| expectedCustomer = custRegion.get(custId); |
| assertNull(expectedCustomer); |
| |
| // Perform two more puts and a commit |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| CustId custIdThree = new CustId(3); |
| Customer customerThree = new Customer("name3", "addr3"); |
| custRegion.put(custIdTwo, customerTwo); |
| custRegion.put(custIdThree, customerThree); |
| |
| mgr.commit(); |
| mgr.begin(); |
| |
| // Verify data |
| assertEquals(2, custRegion.size()); |
| assertTrue(custRegion.containsKey(custIdTwo)); |
| assertTrue(custRegion.containsKey(custIdThree)); |
| assertEquals(customerTwo, custRegion.get(custIdTwo)); |
| assertEquals(customerThree, custRegion.get(custIdThree)); |
| |
| // Perform one more put but don't commit |
| custRegion.put(custIdOne, customerOne); |
| |
| // Verify data |
| assertEquals(3, custRegion.size()); |
| assertTrue(custRegion.containsKey(custIdOne)); |
| assertEquals(customerOne, custRegion.get(custIdOne)); |
| |
| mgr.commit(); |
| |
| return null; |
| } |
| |
| }); |
| } |
| |
| @Test |
| public void testTransactionalPutOnPartitionedRegion() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createPR(new VM[] {server1, server2, server3}); |
| execute(server1, new SerializableCallable() { |
| |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| LogWriter logger = getGemfireCache().getLogger(); |
| |
| mgr.begin(); |
| logger.fine("TEST:Commit-1"); |
| mgr.commit(); |
| |
| mgr.begin(); |
| Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER_PR); |
| CustId custId = new CustId(1); |
| Customer expectedCustomer = custRegion.get(custId); |
| assertNull(expectedCustomer); |
| |
| // Perform a put |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| logger.fine("TEST:Put-1"); |
| custRegion.put(custIdOne, customerOne); |
| |
| // Rollback the transaction |
| logger.fine("TEST:Rollback-1"); |
| mgr.rollback(); |
| |
| mgr.begin(); |
| // Verify that the entry is rolled back |
| expectedCustomer = custRegion.get(custId); |
| assertNull(expectedCustomer); |
| |
| // Perform two more puts and a commit |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| CustId custIdThree = new CustId(3); |
| Customer customerThree = new Customer("name3", "addr3"); |
| logger.fine("TEST:Put-2"); |
| custRegion.put(custIdTwo, customerTwo); |
| logger.fine("TEST:Put-3"); |
| custRegion.put(custIdThree, customerThree); |
| |
| logger.fine("TEST:Commit-2"); |
| mgr.commit(); |
| mgr.begin(); |
| |
| // Verify data |
| assertEquals(2, custRegion.size()); |
| assertTrue(custRegion.containsKey(custIdTwo)); |
| assertTrue(custRegion.containsKey(custIdThree)); |
| assertEquals(customerTwo, custRegion.get(custIdTwo)); |
| assertEquals(customerThree, custRegion.get(custIdThree)); |
| |
| // Perform one more put but don't commit |
| logger.fine("TEST:Put-4"); |
| Customer customerOneMod = new Customer("name1", "addr11"); |
| |
| custRegion.put(custIdOne, customerOneMod); |
| |
| // Verify data |
| assertEquals(3, custRegion.size()); |
| assertTrue(custRegion.containsKey(custIdOne)); |
| assertEquals(customerOneMod, custRegion.get(custIdOne)); |
| logger.fine("TEST:Commit-3"); |
| mgr.commit(); |
| |
| return null; |
| } |
| |
| }); |
| } |
| |
| @SuppressWarnings("serial") |
| @Test |
| public void testCommitOnPartitionedAndReplicatedRegions() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| createPR(new VM[] {server1, server2, server3}); |
| createRR(new VM[] {server1, server2, server3}); |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| Region<CustId, Customer> rrRegion = getCache().getRegion(CUSTOMER_RR); |
| Region<CustId, Customer> prRegion = getCache().getRegion(CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| prRegion.put(custIdOne, customerOne); |
| rrRegion.put(custIdOne, customerOne); |
| |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| rrRegion.put(custIdTwo, customerTwo); |
| |
| mgr.commit(); |
| |
| // Verify |
| assertEquals(2, rrRegion.size()); |
| assertTrue(rrRegion.containsKey(custIdOne)); |
| assertTrue(rrRegion.containsKey(custIdTwo)); |
| assertEquals(customerOne, rrRegion.get(custIdOne)); |
| assertEquals(customerTwo, rrRegion.get(custIdTwo)); |
| |
| assertEquals(1, prRegion.size()); |
| assertTrue(prRegion.containsKey(custIdOne)); |
| assertEquals(customerOne, rrRegion.get(custIdOne)); |
| |
| return null; |
| } |
| |
| }); |
| } |
| |
| @Test |
| public void testGetIsolated() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| createPR(new VM[] {server1, server2, server3}); |
| createRR(new VM[] {server1, server2, server3}); |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| mgr.commit(); |
| mgr.begin(); |
| Region<CustId, Customer> custPR = getCache().getRegion(CUSTOMER_PR); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| custPR.put(custIdOne, customerOne); |
| |
| Region<CustId, Customer> replicatedRegion = getCache().getRegion(CUSTOMER_RR); |
| replicatedRegion.put(custIdOne, customerOne); |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| replicatedRegion.put(custIdTwo, customerTwo); |
| |
| // Verify before commit |
| assertEquals(2, replicatedRegion.size()); |
| assertEquals(customerOne, replicatedRegion.get(custIdOne)); |
| assertEquals(customerTwo, replicatedRegion.get(custIdTwo)); |
| |
| // Perform commit |
| mgr.commit(); |
| |
| // Verify after commit |
| assertEquals(2, replicatedRegion.size()); |
| assertEquals(customerOne, replicatedRegion.get(custIdOne)); |
| assertEquals(customerTwo, replicatedRegion.get(custIdTwo)); |
| |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testCommitAndRollback() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| createPR(new VM[] {server1, server2, server3}); |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| // mgr.begin(); |
| Region<CustId, Customer> custPR = getCache().getRegion(CUSTOMER_PR); |
| for (int i = 0; i < 1000; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("name" + i, "addr" + i); |
| mgr.begin(); |
| custPR.put(custId, customer); |
| if (i % 2 == 0) { |
| mgr.commit(); |
| } else { |
| mgr.rollback(); |
| } |
| } |
| // Verify number of puts |
| assertEquals(500, custPR.size()); |
| Set<Region.Entry<?, ?>> entries = custPR.entrySet(false); |
| assertEquals(500, entries.size()); |
| |
| return null; |
| } |
| }); |
| } |
| |
| /* |
| * We create 2 partitioned regions one on each server and have a third node as accessor and fire |
| * transactional operations on it. |
| */ |
| @Test |
| public void testNonColocatedPutByPartitioning() { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); // datastore |
| VM server2 = host.getVM(1); // datastore |
| VM server3 = host.getVM(2); // accessor |
| |
| final String CUSTOMER_PR1 = "CUSTOMER_PR1"; |
| final String CUSTOMER_PR2 = "CUSTOMER_PR2"; |
| |
| // Create CUSTOMER_PR1 on server1 |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver1")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion(CUSTOMER_PR1, af.create()); |
| return null; |
| } |
| }); |
| |
| // Create CUSTOMER_PR2 on server2 |
| execute(server2, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver2")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion(CUSTOMER_PR2, af.create()); |
| return null; |
| } |
| }); |
| |
| // Create both the regions on server3 (accessor) |
| execute(server3, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(0) // since this is an accessor |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver1")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion(CUSTOMER_PR1, af.create()); |
| |
| return null; |
| } |
| }); |
| execute(server3, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(0) // since this is an accessor |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver2")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion(CUSTOMER_PR2, af.create()); |
| return null; |
| } |
| }); |
| |
| // Now perform tx ops on accessor |
| execute(server3, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| |
| Region<CustId, Customer> custPR1 = getCache().getRegion(CUSTOMER_PR1); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| custPR1.put(custIdOne, customerOne); |
| |
| Region<CustId, Customer> custPR2 = getCache().getRegion(CUSTOMER_PR2); |
| custPR2.put(custIdOne, customerOne); |
| |
| mgr.commit(); |
| |
| // Verify |
| assertEquals(1, custPR1.size()); |
| assertEquals(1, custPR2.size()); |
| return null; |
| } |
| }); |
| |
| // Verify on one of the servers |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> custPR1 = getCache().getRegion(CUSTOMER_PR1); |
| assertEquals(1, custPR1.size()); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| assertEquals(customerOne, custPR1.get(custIdOne)); |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testTransactionalKeyBasedUpdates() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| createPR(new VM[] {server1, server2, server3}); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| // mgr.begin(); |
| LogWriter logger = getGemfireCache().getLogger(); |
| |
| Region<CustId, Customer> custPR = getCache().getRegion(CUSTOMER_PR); |
| for (int i = 1; i <= 2; i++) { |
| mgr.begin(); |
| logger.fine("TEST:PUT-" + i); |
| custPR.put(new CustId(i), new Customer("name" + i, "addr" + i)); |
| logger.fine("TEST:COMMIT-" + i); |
| mgr.commit(); |
| } |
| |
| // Updates |
| for (int i = 1; i <= 2; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = custPR.get(custId); |
| assertNotNull(customer); |
| mgr.begin(); |
| logger.fine("TEST:UPDATE-" + i); |
| custPR.put(custId, new Customer("name" + i * 2, "addr" + i * 2)); |
| logger.fine("TEST:UPDATED-" + i + "=" + custId + "," + custPR.get(custId)); |
| logger.fine("TEST:UPDATE COMMIT-" + i); |
| mgr.commit(); |
| logger.fine("TEST:POSTCOMMIT-" + i + "=" + custId + "," + custPR.get(custId)); |
| } |
| // Verify |
| for (int i = 1; i <= 2; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = custPR.get(custId); |
| assertNotNull(customer); |
| logger.fine("TEST:VERIFYING-" + i); |
| assertEquals(new Customer("name" + i * 2, "addr" + i * 2), customer); |
| } |
| |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testTransactionalKeyBasedDestroys_PR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| createPR(new VM[] {server1, server2, server3}); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| // mgr.begin(); |
| |
| Region<CustId, Customer> custPR = getCache().getRegion(CUSTOMER_PR); |
| for (int i = 1; i <= 1000; i++) { |
| mgr.begin(); |
| custPR.put(new CustId(i), new Customer("name" + i, "addr" + i)); |
| mgr.commit(); |
| } |
| |
| // Destroys |
| for (int i = 1; i <= 100; i++) { |
| CustId custId = new CustId(i); |
| mgr.begin(); |
| Object customerRemoved = custPR.remove(custId); |
| // Removing this assertion since in case of distributed destroys the |
| // value will not be returned. |
| // assertNotNull(customerRemoved); |
| mgr.commit(); |
| } |
| |
| // Verify |
| for (int i = 1; i <= 100; i++) { |
| CustId custId = new CustId(1); |
| Customer customer = custPR.get(custId); |
| assertNull(customer); |
| } |
| |
| assertEquals(900, custPR.size()); |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testTransactionalKeyBasedDestroys_RR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| createRR(new VM[] {server1, server2, server3}); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| // mgr.begin(); |
| |
| Region<CustId, Customer> custRR = getCache().getRegion(CUSTOMER_RR); |
| for (int i = 1; i <= 1000; i++) { |
| mgr.begin(); |
| custRR.put(new CustId(i), new Customer("name" + i, "addr" + i)); |
| mgr.commit(); |
| } |
| |
| // Destroys |
| for (int i = 1; i <= 100; i++) { |
| CustId custId = new CustId(i); |
| mgr.begin(); |
| Object customerRemoved = custRR.remove(custId); |
| assertNotNull(customerRemoved); |
| mgr.commit(); |
| } |
| |
| // Verify |
| for (int i = 1; i <= 100; i++) { |
| CustId custId = new CustId(1); |
| Customer customer = custRR.get(custId); |
| assertNull(customer); |
| } |
| |
| assertEquals(900, custRR.size()); |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testTransactionalUpdates() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createPR(new VM[] {server2, server3}); |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| createPR(false, 0, null); |
| |
| Region<CustId, Customer> custPR = getCache().getRegion(CUSTOMER_PR); |
| for (int i = 1; i <= 200; i++) { |
| custPR.put(new CustId(i), new Customer("name" + i, "addr" + i)); |
| } |
| |
| assertEquals(200, custPR.size()); |
| |
| mgr.rollback(); |
| // mgr.commit(); |
| mgr.begin(); |
| assertEquals(0, custPR.size()); |
| |
| mgr.commit(); |
| mgr.begin(); |
| |
| for (int i = 1; i <= 200; i++) { |
| custPR.put(new CustId(i), new Customer("name" + i, "addr" + i)); |
| } |
| |
| mgr.commit(); |
| // mgr.begin(); |
| for (int i = 1; i <= 200; i++) { |
| mgr.begin(); |
| custPR.put(new CustId(i), new Customer("name" + i * 2, "addr" + i * 2)); |
| mgr.commit(); |
| } |
| mgr.begin(); |
| mgr.rollback(); |
| |
| assertEquals(200, custPR.size()); |
| |
| for (int i = 1; i <= 200; i++) { |
| assertEquals(new Customer("name" + i * 2, "addr" + i * 2), custPR.get(new CustId(i))); |
| } |
| |
| return null; |
| } |
| |
| }); |
| } |
| |
| @Test |
| public void testPutAllWithTransactions() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createRegions(new VM[] {server1, server2, server3}); |
| |
| execute(new VM[] {server1, server2, server3}, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver1")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion("NONCOLOCATED_PR", af.create()); |
| return null; |
| } |
| }); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region custRegion = getCache().getRegion(CUSTOMER_PR); |
| Region orderRegion = getCache().getRegion(ORDER_PR); |
| |
| Map custMap = new HashMap(); |
| Map orderMap = new HashMap(); |
| for (int i = 0; i < 5; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custMap.put(custId, customer); |
| orderMap.put(orderId, order); |
| } |
| |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| custRegion.putAll(custMap); |
| orderRegion.putAll(orderMap); |
| mgr.commit(); |
| |
| mgr.begin(); |
| assertEquals(5, custRegion.size()); |
| assertEquals(5, orderRegion.size()); |
| |
| custMap = new HashMap(); |
| orderMap = new HashMap(); |
| for (int i = 5; i < 10; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custMap.put(custId, customer); |
| orderMap.put(orderId, order); |
| } |
| custRegion.putAll(custMap); |
| orderRegion.putAll(orderMap); |
| mgr.rollback(); |
| |
| mgr.begin(); |
| assertEquals(5, custRegion.size()); |
| assertEquals(5, orderRegion.size()); |
| |
| custRegion.putAll(custMap); |
| orderRegion.putAll(orderMap); |
| |
| assertEquals(10, custRegion.size()); |
| assertEquals(10, orderRegion.size()); |
| |
| // Verify operations involving non colocated PR |
| Map map = new HashMap(); |
| custMap.clear(); |
| for (int i = 10; i < 15; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custMap.put(custId, customer); |
| map.put(custId, customer); |
| } |
| custRegion.putAll(custMap); |
| Region nonColocatedRegion = getCache().getRegion("NONCOLOCATED_PR"); |
| nonColocatedRegion.putAll(orderMap); |
| |
| |
| mgr.commit(); |
| mgr.begin(); |
| assertEquals(15, custRegion.size()); |
| assertEquals(5, nonColocatedRegion.size()); |
| |
| custMap.clear(); |
| map.clear(); |
| for (int i = 15; i < 20; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custMap.put(custId, customer); |
| map.put(custId, customer); |
| } |
| custRegion.putAll(custMap); |
| nonColocatedRegion.putAll(orderMap); |
| |
| mgr.rollback(); |
| assertEquals(15, custRegion.size()); |
| assertEquals(5, nonColocatedRegion.size()); |
| |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testRemoveAllWithTransactions() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createRegions(new VM[] {server1, server2, server3}); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region custRegion = getCache().getRegion(CUSTOMER_PR); |
| Region orderRegion = getCache().getRegion(ORDER_PR); |
| |
| Map custMap = new HashMap(); |
| Map orderMap = new HashMap(); |
| for (int i = 0; i < 15; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custMap.put(custId, customer); |
| orderMap.put(orderId, order); |
| } |
| |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| custRegion.putAll(custMap); |
| orderRegion.putAll(orderMap); |
| mgr.commit(); |
| |
| mgr.begin(); |
| assertEquals(15, custRegion.size()); |
| assertEquals(15, orderRegion.size()); |
| |
| custMap = new HashMap(); |
| orderMap = new HashMap(); |
| for (int i = 5; i < 10; i++) { |
| CustId custId = new CustId(i); |
| Customer customer = new Customer("customer" + i, "address" + i); |
| OrderId orderId = new OrderId(i, custId); |
| Order order = new Order("order" + i); |
| custMap.put(custId, customer); |
| orderMap.put(orderId, order); |
| } |
| custRegion.removeAll(custMap.keySet()); |
| orderRegion.removeAll(orderMap.keySet()); |
| mgr.rollback(); |
| |
| mgr.begin(); |
| assertEquals(15, custRegion.size()); |
| assertEquals(15, orderRegion.size()); |
| |
| custRegion.removeAll(custMap.keySet()); |
| orderRegion.removeAll(orderMap.keySet()); |
| |
| assertEquals(10, custRegion.size()); |
| assertEquals(10, orderRegion.size()); |
| mgr.commit(); |
| |
| assertEquals(10, custRegion.size()); |
| assertEquals(10, orderRegion.size()); |
| |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testTxWithSingleDataStore() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); // datastore |
| |
| final String CUSTOMER_PR1 = "CUSTOMER_PR1"; |
| final String CUSTOMER_PR2 = "CUSTOMER_PR2"; |
| |
| // Create CUSTOMER_PR1 on server1 |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver1")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion(CUSTOMER_PR1, af.create()); |
| return null; |
| } |
| }); |
| |
| // Create CUSTOMER_PR2 on server2 |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| AttributesFactory af = new AttributesFactory(); |
| af.setConcurrencyChecksEnabled(getConcurrencyChecksEnabled()); |
| af.setPartitionAttributes(new PartitionAttributesFactory<CustId, Customer>() |
| .setTotalNumBuckets(4).setLocalMaxMemory(1) |
| .setPartitionResolver(new CustomerIDPartitionResolver("resolver2")) |
| .setRedundantCopies(0).create()); |
| getCache().createRegion(CUSTOMER_PR2, af.create()); |
| return null; |
| } |
| }); |
| |
| // Now perform tx ops on accessor |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| |
| Region<CustId, Customer> custPR1 = getCache().getRegion(CUSTOMER_PR1); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| custPR1.put(custIdOne, customerOne); |
| |
| Region<CustId, Customer> custPR2 = getCache().getRegion(CUSTOMER_PR2); |
| custPR2.put(custIdOne, customerOne); |
| |
| mgr.commit(); |
| |
| // Verify |
| assertEquals(1, custPR1.size()); |
| assertEquals(1, custPR2.size()); |
| return null; |
| } |
| }); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> custPR1 = getCache().getRegion(CUSTOMER_PR1); |
| assertEquals(1, custPR1.size()); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| assertEquals(customerOne, custPR1.get(custIdOne)); |
| return null; |
| } |
| }); |
| } |
| |
| @Test |
| public void testMultipleOpsOnSameKeyInTx() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createPR(new VM[] {server1, server2, server3}); |
| execute(server1, new SerializableCallable() { |
| |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| mgr.commit(); |
| |
| mgr.begin(); |
| Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER_PR); |
| CustId custId = new CustId(1); |
| Customer expectedCustomer = custRegion.get(custId); |
| assertNull(expectedCustomer); |
| |
| // Perform a put |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| custRegion.put(custIdOne, customerOne); |
| |
| // Rollback the transaction |
| mgr.rollback(); |
| |
| mgr.begin(); |
| // Verify that the entry is rolled back |
| expectedCustomer = custRegion.get(custId); |
| assertNull(expectedCustomer); |
| |
| // Add more data |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| CustId custIdThree = new CustId(3); |
| Customer customerThree = new Customer("name3", "addr3"); |
| custRegion.put(custIdTwo, customerTwo); |
| custRegion.put(custIdThree, customerThree); |
| mgr.commit(); |
| |
| mgr.begin(); |
| // Verify data |
| assertEquals(2, custRegion.size()); |
| assertTrue(custRegion.containsKey(custIdTwo)); |
| assertTrue(custRegion.containsKey(custIdThree)); |
| assertEquals(customerTwo, custRegion.get(custIdTwo)); |
| assertEquals(customerThree, custRegion.get(custIdThree)); |
| |
| // Update the values for the same keys multiple times |
| custRegion.put(custIdOne, new Customer("name1_mod1", "addr1_mod1")); |
| custRegion.put(custIdTwo, new Customer("name2_mod1", "addr2_mod1")); |
| custRegion.put(custIdOne, new Customer("name1_mod2", "addr1_mod2")); |
| custRegion.put(custIdOne, new Customer("name1_mod3", "addr1_mod3")); |
| custRegion.put(custIdTwo, new Customer("name2_mod2", "addr2_mod2")); |
| |
| assertEquals(3, custRegion.size()); |
| mgr.commit(); |
| |
| assertEquals(3, custRegion.size()); |
| Customer c = custRegion.get(custIdOne); |
| return null; |
| } |
| }); |
| |
| } |
| |
| |
| /* |
| * Test to reproduce a scenario where: 1. On primary, the tx op is applied first followed by |
| * non-tx 2. On secondary, non-tx op is applied first followed by tx. |
| */ |
| @Ignore |
| @Test |
| public void testConcurrentTXAndNonTXOperations() throws Exception { |
| Host host = Host.getHost(0); |
| final VM server1 = host.getVM(0); |
| final VM server2 = host.getVM(1); |
| |
| createPersistentPR(new VM[] {server1}); |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| prRegion.put(custIdOne, customerOne); |
| |
| BucketRegion br = ((PartitionedRegion) prRegion).getBucketRegion(custIdOne); |
| |
| String primaryMember = br.getBucketAdvisor().getPrimary().toString(); |
| getGemfireCache().getLogger().fine("TEST:PRIMARY:" + primaryMember); |
| |
| String memberId = getGemfireCache().getDistributedSystem().getMemberId(); |
| getGemfireCache().getLogger().fine("TEST:MEMBERID:" + memberId); |
| |
| return null; |
| } |
| }); |
| |
| createPersistentPR(new VM[] {server2}); |
| |
| Boolean isPrimary = (Boolean) execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| CustId custIdOne = new CustId(1); |
| BucketRegion br = ((PartitionedRegion) prRegion).getBucketRegion(custIdOne); |
| |
| String primaryMember = br.getBucketAdvisor().getPrimary().toString(); |
| getGemfireCache().getLogger().fine("TEST:PRIMARY:" + primaryMember); |
| |
| String memberId = getGemfireCache().getDistributedSystem().getMemberId(); |
| getGemfireCache().getLogger().fine("TEST:MEMBERID:" + memberId); |
| |
| return memberId.equals(primaryMember); |
| } |
| }); |
| |
| final VM primary = isPrimary.booleanValue() ? server1 : server2; |
| final VM secondary = !isPrimary.booleanValue() ? server1 : server2; |
| |
| System.out.println("TEST:SERVER-1:VM-" + server1.getId()); |
| System.out.println("TEST:SERVER-2:VM-" + server2.getId()); |
| System.out.println("TEST:PRIMARY=VM-" + primary.getId()); |
| System.out.println("TEST:SECONDARY=VM-" + secondary.getId()); |
| |
| class WaitRelease implements Runnable { |
| CountDownLatch cdl; |
| String op; |
| |
| public WaitRelease(CountDownLatch cdl, String member) { |
| this.cdl = cdl; |
| } |
| |
| @Override |
| public void run() { |
| try { |
| GemFireCacheImpl.getExisting().getLogger().fine("TEST:TX WAITING - " + op); |
| cdl.await(); |
| GemFireCacheImpl.getExisting().getLogger().fine("TEST:TX END WAITING"); |
| } catch (InterruptedException e) { |
| } |
| } |
| |
| public void release() { |
| GemFireCacheImpl.getExisting().getLogger().fine("TEST:TX COUNTDOWN - " + op); |
| cdl.countDown(); |
| } |
| } |
| |
| // Install TX hook |
| SerializableCallable txHook = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CountDownLatch cdl = new CountDownLatch(1); |
| DistTXState.internalBeforeApplyChanges = new WaitRelease(cdl, "TX OP"); |
| return null; |
| } |
| }; |
| |
| execute(secondary, txHook); |
| |
| |
| // Install non-TX hook |
| SerializableCallable nontxHook = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CountDownLatch cdl = new CountDownLatch(1); |
| DistTXState.internalBeforeNonTXBasicPut = new WaitRelease(cdl, "NON TX OP"); |
| return null; |
| } |
| }; |
| |
| // Install the wait-release hook on the secondary |
| execute(secondary, nontxHook); |
| |
| |
| // Start a tx operation on primary |
| |
| execute(primary, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| // The reason this is run in a separate thread instead of controller thread |
| // is that this is going to block because the secondary is going to wait. |
| new Thread() { |
| @Override |
| public void run() { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| getGemfireCache().getLogger().fine("TEST:DISTTX=" + mgr.isDistributed()); |
| mgr.begin(); |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1_tx", "addr1"); |
| getGemfireCache().getLogger().fine("TEST:TX UPDATE"); |
| prRegion.put(custIdOne, customerOne); |
| getGemfireCache().getLogger().fine("TEST:TX COMMIT"); |
| mgr.commit(); |
| } |
| }.start(); |
| return null; |
| } |
| }); |
| |
| // Let the TX op be applied on primary first |
| Thread.currentThread().sleep(200); |
| |
| // Perform a non-tx op on the same key on primary |
| execute(primary, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1_nontx", "addr1"); |
| getGemfireCache().getLogger().fine("TEST:TX NONTXUPDATE"); |
| prRegion.put(custIdOne, customerOne); |
| return null; |
| } |
| }); |
| |
| |
| // Wait for a few milliseconds |
| Thread.currentThread().sleep(200); |
| |
| // Release the waiting non-tx op first, on secondary |
| execute(secondary, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Runnable r = DistTXState.internalBeforeNonTXBasicPut; |
| assert (r != null && r instanceof WaitRelease); |
| WaitRelease e = (WaitRelease) r; |
| e.release(); |
| return null; |
| } |
| }); |
| |
| // Now release the waiting commit on secondary |
| execute(secondary, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Runnable r = DistTXState.internalBeforeApplyChanges; |
| assert (r != null && r instanceof WaitRelease); |
| WaitRelease e = (WaitRelease) r; |
| e.release(); |
| return null; |
| } |
| }); |
| |
| // Verify region and entry versions on primary and secondary |
| SerializableCallable verifyPrimary = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custId = new CustId(1); |
| Customer customer = prRegion.get(custId); |
| |
| BucketRegion br = ((PartitionedRegion) prRegion).getBucketRegion(custId); |
| RegionEntry re = br.getRegionEntry(custId); |
| |
| getGemfireCache().getLogger().fine("TEST:TX PRIMARY CUSTOMER=" + customer); |
| |
| getGemfireCache().getLogger() |
| .fine("TEST:TX PRIMARY REGION VERSION=" + re.getVersionStamp().getRegionVersion()); |
| getGemfireCache().getLogger() |
| .fine("TEST:TX PRIMARY ENTRY VERSION=" + re.getVersionStamp().getEntryVersion()); |
| return null; |
| } |
| }; |
| execute(primary, verifyPrimary); |
| SerializableCallable verifySecondary = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custId = new CustId(1); |
| Customer customer = prRegion.get(custId); |
| |
| BucketRegion br = ((PartitionedRegion) prRegion).getBucketRegion(custId); |
| RegionEntry re = br.getRegionEntry(custId); |
| |
| getGemfireCache().getLogger().fine("TEST:TX SECONDARY CUSTOMER=" + customer); |
| |
| getGemfireCache().getLogger() |
| .fine("TEST:TX SECONDARY REGION VERSION=" + re.getVersionStamp().getRegionVersion()); |
| getGemfireCache().getLogger() |
| .fine("TEST:TX SECONDARY ENTRY VERSION=" + re.getVersionStamp().getEntryVersion()); |
| return null; |
| } |
| }; |
| |
| execute(secondary, verifySecondary); |
| } |
| |
| @Test |
| public void testBasicDistributedTX() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| createPersistentPR(new VM[] {server1, server2}); |
| execute(server2, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| getGemfireCache().getLogger().fine("TEST:DISTTX=" + mgr.isDistributed()); |
| getGemfireCache().getLogger().fine("TEST:TX BEGIN"); |
| mgr.begin(); |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| getGemfireCache().getLogger().fine("TEST:TX PUT 1"); |
| prRegion.put(custIdOne, customerOne); |
| |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| getGemfireCache().getLogger().fine("TEST:TX PUT 2"); |
| prRegion.put(custIdTwo, customerTwo); |
| |
| getGemfireCache().getLogger().fine("TEST:TX COMMIT"); |
| mgr.commit(); |
| return null; |
| } |
| }); |
| |
| } |
| |
| @Test |
| public void testRegionAndEntryVersionsPR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| createPersistentPR(new VM[] {server1, server2}); |
| execute(server2, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| getGemfireCache().getLogger().fine("TEST:DISTTX=" + mgr.isDistributed()); |
| getGemfireCache().getLogger().fine("TEST:TX BEGIN"); |
| mgr.begin(); |
| |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| getGemfireCache().getLogger().fine("TEST:TX PUT 1"); |
| prRegion.put(custIdOne, customerOne); |
| |
| BucketRegion br = ((PartitionedRegion) prRegion).getBucketRegion(custIdOne); |
| |
| assertEquals(0L, br.getVersionVector().getCurrentVersion()); |
| getGemfireCache().getLogger().fine("TEST:TX COMMIT 1"); |
| mgr.commit(); |
| |
| // Verify region version on the region |
| assertEquals(1L, br.getVersionVector().getCurrentVersion()); |
| |
| RegionEntry re = br.getRegionEntry(custIdOne); |
| getGemfireCache().getLogger().fine("TEST:VERSION-STAMP:" + re.getVersionStamp()); |
| |
| // Verify region version on the region entry |
| assertEquals(1L, re.getVersionStamp().getRegionVersion()); |
| |
| // Verify entry version |
| assertEquals(1, re.getVersionStamp().getEntryVersion()); |
| |
| mgr.begin(); |
| prRegion.put(custIdOne, new Customer("name1_1", "addr1")); |
| |
| getGemfireCache().getLogger().fine("TEST:TX COMMIT 2"); |
| |
| assertEquals(1L, br.getVersionVector().getCurrentVersion()); |
| mgr.commit(); |
| |
| // Verify region version on the region |
| assertEquals(2L, br.getVersionVector().getCurrentVersion()); |
| |
| re = br.getRegionEntry(custIdOne); |
| getGemfireCache().getLogger().fine("TEST:VERSION-STAMP:" + re.getVersionStamp()); |
| |
| // Verify region version on the region entry |
| assertEquals(2L, re.getVersionStamp().getRegionVersion()); |
| |
| // Verify entry version |
| assertEquals(2, re.getVersionStamp().getEntryVersion()); |
| return null; |
| } |
| |
| }); |
| |
| execute(server1, new SerializableCallable() { |
| |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> prRegion = getCache().getRegion(PERSISTENT_CUSTOMER_PR); |
| CustId custIdOne = new CustId(1); |
| BucketRegion br = ((PartitionedRegion) prRegion).getBucketRegion(custIdOne); |
| |
| // Verify region version on the region |
| assertEquals(2L, br.getVersionVector().getCurrentVersion()); |
| |
| // Verify region version ont the region entry |
| RegionEntry re = br.getRegionEntry(custIdOne); |
| assertEquals(2L, re.getVersionStamp().getRegionVersion()); |
| |
| // Verify entry version |
| assertEquals(2, re.getVersionStamp().getEntryVersion()); |
| return null; |
| } |
| |
| }); |
| |
| } |
| |
| @Test |
| public void testRegionAndEntryVersionsRR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| createRR(new VM[] {server1, server2}); |
| execute(server2, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| getGemfireCache().getLogger().fine("TEST:DISTTX=" + mgr.isDistributed()); |
| getGemfireCache().getLogger().fine("TEST:TX BEGIN"); |
| mgr.begin(); |
| |
| Region<CustId, Customer> region = getCache().getRegion(CUSTOMER_RR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| getGemfireCache().getLogger().fine("TEST:TX PUT 1"); |
| region.put(custIdOne, customerOne); |
| |
| LocalRegion lr = (LocalRegion) region; |
| |
| assertEquals(0L, lr.getVersionVector().getCurrentVersion()); |
| getGemfireCache().getLogger().fine("TEST:TX COMMIT 1"); |
| mgr.commit(); |
| |
| // Verify region version on the region |
| assertEquals(1L, lr.getVersionVector().getCurrentVersion()); |
| |
| RegionEntry re = lr.getRegionEntry(custIdOne); |
| getGemfireCache().getLogger().fine("TEST:VERSION-STAMP:" + re.getVersionStamp()); |
| |
| // Verify region version on the region entry |
| assertEquals(1L, re.getVersionStamp().getRegionVersion()); |
| |
| // Verify entry version |
| assertEquals(1, re.getVersionStamp().getEntryVersion()); |
| |
| mgr.begin(); |
| region.put(custIdOne, new Customer("name1_1", "addr1")); |
| |
| getGemfireCache().getLogger().fine("TEST:TX COMMIT 2"); |
| |
| assertEquals(1L, lr.getVersionVector().getCurrentVersion()); |
| mgr.commit(); |
| |
| // Verify region version on the region |
| assertEquals(2L, lr.getVersionVector().getCurrentVersion()); |
| |
| re = lr.getRegionEntry(custIdOne); |
| getGemfireCache().getLogger().fine("TEST:VERSION-STAMP:" + re.getVersionStamp()); |
| |
| // Verify region version on the region entry |
| assertEquals(2L, re.getVersionStamp().getRegionVersion()); |
| |
| // Verify entry version |
| assertEquals(2, re.getVersionStamp().getEntryVersion()); |
| return null; |
| } |
| |
| }); |
| |
| execute(server1, new SerializableCallable() { |
| |
| @Override |
| public Object call() throws Exception { |
| Region<CustId, Customer> region = getCache().getRegion(CUSTOMER_RR); |
| CustId custIdOne = new CustId(1); |
| |
| LocalRegion lr = (LocalRegion) region; |
| |
| // Verify region version on the region |
| assertEquals(2L, lr.getVersionVector().getCurrentVersion()); |
| |
| // Verify region version ont the region entry |
| RegionEntry re = lr.getRegionEntry(custIdOne); |
| assertEquals(2L, re.getVersionStamp().getRegionVersion()); |
| |
| // Verify entry version |
| assertEquals(2, re.getVersionStamp().getEntryVersion()); |
| return null; |
| } |
| |
| }); |
| |
| } |
| |
| |
| @Test |
| public void testTxWorksWithNewNodeJoining() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| |
| createPR(new VM[] {server1, server2}); |
| |
| class Ops extends SerializableCallable { |
| private boolean flag = false; |
| |
| public void setWaitFlag(boolean value) { |
| flag = value; |
| } |
| |
| @Override |
| public Object call() throws Exception { |
| |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| Region<CustId, Customer> prRegion = getCache().getRegion(CUSTOMER_PR); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| prRegion.put(custIdOne, customerOne); |
| |
| |
| // Install the hook at this point |
| TestObserver o = TestObserver.getInstance(); |
| o.setFlag(true); |
| |
| while (o.getFlag()) { |
| Thread.currentThread().sleep(1000); |
| } |
| |
| mgr.commit(); |
| |
| // Verify |
| assertEquals(1, prRegion.size()); |
| assertTrue(prRegion.containsKey(custIdOne)); |
| |
| return null; |
| } |
| } |
| |
| server1.invokeAsync(() -> new Ops().call()); |
| |
| // Now create cache on the third server and let it join the distributed system. |
| createPR(new VM[] {server3}); |
| |
| // Let the original thread move on by signalling the flag |
| |
| execute(server1, new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| TestObserver o = TestObserver.getInstance(); |
| o.setFlag(false); |
| return null; |
| } |
| }); |
| } |
| |
| public static class TestObserver { |
| private static final TestObserver singleInstance = new TestObserver(); |
| private static boolean flag = false; |
| |
| public static TestObserver getInstance() { |
| return singleInstance; |
| } |
| |
| public static void setFlag(boolean val) { |
| flag = val; |
| } |
| |
| public boolean getFlag() { |
| return flag; |
| } |
| } |
| |
| |
| private class TxOps_Conflicts extends SerializableCallable { |
| |
| final String regionName; |
| |
| public TxOps_Conflicts(String regionName) { |
| this.regionName = regionName; |
| } |
| |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| |
| // Perform a put |
| Region<CustId, Customer> custRegion = getCache().getRegion(this.regionName); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| CustId custIdThree = new CustId(3); |
| Customer customerThree = new Customer("name3", "addr3"); |
| custRegion.put(custIdOne, customerOne); |
| custRegion.put(custIdTwo, customerTwo); |
| custRegion.put(custIdThree, customerThree); |
| |
| // spawn a new thread modify and custIdOne in another tx |
| // so that outer thread fails |
| class TxThread extends Thread { |
| @Override |
| public void run() { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr11"); |
| Region<CustId, Customer> custRegion = getCache().getRegion(regionName); |
| custRegion.put(custIdOne, customerOne); |
| mgr.commit(); |
| } |
| } |
| |
| TxThread txThread = new TxThread(); |
| txThread.start(); |
| txThread.join(); // let the tx commit |
| |
| try { |
| mgr.commit(); |
| fail("this test should have failed with CommitConflictException"); |
| // [DISTTX] TODO after conflict detection either |
| // CommitIncompleteException or CommitConflictException is thrown. |
| // Should it always be CommitConflictException? |
| } catch (CommitIncompleteException cie) { |
| } catch (CommitConflictException ce) { |
| } |
| |
| // verify data |
| assertEquals(new Customer("name1", "addr11"), custRegion.get(custIdOne)); |
| assertEquals(null, custRegion.get(custIdTwo)); |
| assertEquals(null, custRegion.get(custIdThree)); |
| |
| // clearing the region |
| custRegion.remove(custIdOne); |
| return null; |
| } |
| } |
| |
| |
| /* |
| * Start two concurrent transactions that put same entries. Make sure that conflict is detected at |
| * the commit time. |
| */ |
| @Test |
| public void testCommitConflicts_PR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| VM accessor = host.getVM(3); |
| |
| createPRwithRedundanyCopies(new VM[] {server1, server2, server3}, 1); |
| createPRonAccessor(accessor, 1); |
| |
| server1.invoke(new TxOps_Conflicts(CUSTOMER_PR)); |
| |
| // test thru accessor as well |
| accessor.invoke(new TxOps_Conflicts(CUSTOMER_PR)); |
| } |
| |
| /* |
| * Start two concurrent transactions that put same entries. Make sure that conflict is detected at |
| * the commit time. |
| */ |
| @Test |
| public void testCommitConflicts_RR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| VM server3 = host.getVM(2); |
| VM accessor = host.getVM(3); |
| |
| createRR(new VM[] {server1, server2, server3}); |
| createRRonAccessor(accessor); |
| |
| server1.invoke(new TxOps_Conflicts(CUSTOMER_RR)); |
| |
| // test thru accessor as well |
| accessor.invoke(new TxOps_Conflicts(CUSTOMER_RR)); |
| } |
| |
| |
| class TxConflictRunnable implements Runnable { |
| final String regionName; |
| |
| public TxConflictRunnable(String regionName) { |
| this.regionName = regionName; |
| } |
| |
| @Override |
| public void run() { |
| // spawn a new thread modify and custIdOne in another tx |
| // so that outer thread fails |
| class TxThread extends Thread { |
| public boolean gotConflict = false; |
| public boolean gotOtherException = false; |
| public Exception ex = new Exception(); |
| |
| @Override |
| public void run() { |
| LogWriterUtils.getLogWriter() |
| .info("Inside TxConflictRunnable.TxThread after aquiring locks"); |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr11"); |
| Region<CustId, Customer> custRegion = getCache().getRegion(regionName); |
| custRegion.put(custIdOne, customerOne); |
| try { |
| mgr.commit(); |
| } catch (CommitConflictException ce) { |
| gotConflict = true; |
| LogWriterUtils.getLogWriter().info("Received exception ", ce); |
| } catch (Exception e) { |
| gotOtherException = true; |
| LogWriterUtils.getLogWriter().info("Received exception ", e); |
| ex.initCause(e); |
| } |
| } |
| } |
| |
| TxThread txThread = new TxThread(); |
| txThread.start(); |
| try { |
| txThread.join(); // let the tx commit |
| } catch (InterruptedException e) { |
| // TODO Auto-generated catch block |
| e.printStackTrace(); |
| } |
| |
| assertTrue("This test should fail with CommitConflictException", txThread.gotConflict); |
| if (txThread.gotOtherException) { |
| Assert.fail("Received unexpected exception ", txThread.ex); |
| } |
| } |
| } |
| |
| private class TxOps_conflicts_after_locks_acquired extends SerializableCallable { |
| |
| final String regionName; |
| |
| public TxOps_conflicts_after_locks_acquired(String regionName) { |
| this.regionName = regionName; |
| } |
| |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| |
| // set up a callback to be invoked after locks are acquired at commit time |
| ((TXStateProxyImpl) ((TXManagerImpl) mgr).getTXState()).forceLocalBootstrap(); |
| TXStateInterface txp = ((TXManagerImpl) mgr).getTXState(); |
| DistTXState tx = (DistTXState) ((TXStateProxyImpl) txp).getRealDeal(null, null); |
| tx.setAfterReservation(new TxConflictRunnable(this.regionName)); // callback |
| |
| // Perform a put |
| Region<CustId, Customer> custRegion = getCache().getRegion(this.regionName); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| CustId custIdThree = new CustId(3); |
| Customer customerThree = new Customer("name3", "addr3"); |
| CustId custIdFour = new CustId(4); |
| Customer customerFour = new Customer("name4", "addr4"); |
| custRegion.put(custIdOne, customerOne); |
| custRegion.put(custIdTwo, customerTwo); |
| custRegion.put(custIdThree, customerThree); |
| custRegion.put(custIdFour, customerFour); |
| |
| // will invoke the callback that spawns a new thread and another |
| // transaction |
| mgr.commit(); |
| |
| // verify data |
| assertEquals(new Customer("name1", "addr1"), custRegion.get(custIdOne)); |
| assertEquals(new Customer("name2", "addr2"), custRegion.get(custIdTwo)); |
| assertEquals(new Customer("name3", "addr3"), custRegion.get(custIdThree)); |
| assertEquals(new Customer("name4", "addr4"), custRegion.get(custIdFour)); |
| |
| return null; |
| } |
| } |
| |
| /* |
| * Start a transaction, at commit time after acquiring locks, start another transaction in a new |
| * thread that modifies same entries as in the earlier transaction. Make sure that conflict is |
| * detected |
| */ |
| @Test |
| public void testCommitConflicts_PR_after_locks_acquired() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| // createPRwithRedundanyCopies(new VM[] { server1, server2 }, 1); |
| |
| createPRwithRedundanyCopies(new VM[] {server1}, 0); |
| |
| server1.invoke(new TxOps_conflicts_after_locks_acquired(CUSTOMER_PR)); |
| } |
| |
| /* |
| * Start a transaction, at commit time after acquiring locks, start another transaction in a new |
| * thread that modifies same entries as in the earlier transaction. Make sure that conflict is |
| * detected |
| */ |
| @Test |
| public void testCommitConflicts_RR_after_locks_acquired() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| createRR(new VM[] {server1, server2}); |
| |
| server1.invoke(new TxOps_conflicts_after_locks_acquired(CUSTOMER_RR)); |
| } |
| |
| |
| class TxRunnable implements Runnable { |
| final String regionName; |
| |
| public TxRunnable(String regionName) { |
| this.regionName = regionName; |
| } |
| |
| @Override |
| public void run() { |
| class TxThread extends Thread { |
| public boolean gotException = false; |
| public Exception ex = new Exception(); |
| |
| @Override |
| public void run() { |
| LogWriterUtils.getLogWriter().info("Inside TxRunnable.TxThread after aquiring locks"); |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| Region<CustId, Customer> custRegion = getCache().getRegion(regionName); |
| for (int i = 11; i <= 20; i++) { |
| custRegion.put(new CustId(i), new Customer("name" + i, "addr" + i)); |
| } |
| try { |
| mgr.commit(); |
| } catch (Exception e) { |
| gotException = true; |
| LogWriterUtils.getLogWriter().info("Received exception ", e); |
| ex.initCause(e); |
| } |
| } |
| } |
| |
| TxThread txThread = new TxThread(); |
| txThread.start(); |
| try { |
| txThread.join(); // let the tx commit |
| } catch (InterruptedException e) { |
| // TODO Auto-generated catch block |
| e.printStackTrace(); |
| } |
| if (txThread.gotException) { |
| Assert.fail("Received exception ", txThread.ex); |
| } |
| } |
| } |
| |
| private class TxOps_no_conflicts extends SerializableCallable { |
| |
| final String regionName; |
| |
| public TxOps_no_conflicts(String regionName) { |
| this.regionName = regionName; |
| } |
| |
| @Override |
| public Object call() throws Exception { |
| CacheTransactionManager mgr = getGemfireCache().getTxManager(); |
| mgr.setDistributed(true); |
| mgr.begin(); |
| |
| // set up a callback to be invoked after locks are acquired at commit time |
| ((TXStateProxyImpl) ((TXManagerImpl) mgr).getTXState()).forceLocalBootstrap(); |
| TXStateInterface txp = ((TXManagerImpl) mgr).getTXState(); |
| DistTXState tx = (DistTXState) ((TXStateProxyImpl) txp).getRealDeal(null, null); |
| tx.setAfterReservation(new TxRunnable(this.regionName)); // callback |
| |
| // Perform a put |
| Region<CustId, Customer> custRegion = getCache().getRegion(this.regionName); |
| |
| CustId custIdOne = new CustId(1); |
| Customer customerOne = new Customer("name1", "addr1"); |
| CustId custIdTwo = new CustId(2); |
| Customer customerTwo = new Customer("name2", "addr2"); |
| CustId custIdThree = new CustId(3); |
| Customer customerThree = new Customer("name3", "addr3"); |
| CustId custIdFour = new CustId(4); |
| Customer customerFour = new Customer("name4", "addr4"); |
| custRegion.put(custIdOne, customerOne); |
| custRegion.put(custIdTwo, customerTwo); |
| custRegion.put(custIdThree, customerThree); |
| custRegion.put(custIdFour, customerFour); |
| |
| // will invoke the callback that spawns a new thread and another |
| // transaction that does puts of 10 entries |
| mgr.commit(); |
| |
| // verify data |
| assertEquals(14, custRegion.size()); |
| |
| return null; |
| } |
| } |
| |
| /* |
| * Start a transaction, at commit time after acquiring locks, start another transaction in a new |
| * thread that modifies different entries Make sure that there is no conflict or exception. |
| */ |
| @Test |
| public void testCommitNoConflicts_PR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| createPRwithRedundanyCopies(new VM[] {server1, server2}, 1); |
| |
| server1.invoke(new TxOps_no_conflicts(CUSTOMER_PR)); |
| } |
| |
| /* |
| * Start a transaction, at commit time after acquiring locks, start another transaction in a new |
| * thread that modifies different entries Make sure that there is no conflict or exception. |
| */ |
| @Test |
| public void testCommitNoConflicts_RR() throws Exception { |
| Host host = Host.getHost(0); |
| VM server1 = host.getVM(0); |
| VM server2 = host.getVM(1); |
| |
| createRR(new VM[] {server1, server2}); |
| |
| server1.invoke(new TxOps_no_conflicts(CUSTOMER_RR)); |
| } |
| |
| |
| |
| } |
