You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by es...@apache.org on 2018/10/30 17:01:45 UTC
[geode] branch feature/GEODE-5953 updated: GEODE-5953: set
transaction target to local member if member host transaction.
This is an automated email from the ASF dual-hosted git repository.
eshu11 pushed a commit to branch feature/GEODE-5953
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/feature/GEODE-5953 by this push:
new 6b97346 GEODE-5953: set transaction target to local member if member host transaction.
6b97346 is described below
commit 6b973466336ee228f7881ec11344359a30a6ac96
Author: eshu <es...@pivotal.io>
AuthorDate: Tue Oct 30 09:58:43 2018 -0700
GEODE-5953: set transaction target to local member if member host transaction.
* to be able to detect primary bucket has moved due to rebalance.
* add debug level logging for transaction.
---
...PartitioningWithTransactionDistributedTest.java | 379 +++++++++++++++++++++
.../apache/geode/internal/cache/TXManagerImpl.java | 16 +-
.../geode/internal/cache/TXStateProxyImpl.java | 5 +-
.../geode/internal/cache/TXManagerImplTest.java | 9 +
.../geode/internal/cache/TXStateProxyImplTest.java | 59 ++++
.../apache/geode/test/dunit/rules/CacheRule.java | 2 +-
6 files changed, 467 insertions(+), 3 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithTransactionDistributedTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithTransactionDistributedTest.java
new file mode 100644
index 0000000..d805192
--- /dev/null
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/fixed/FixedPartitioningWithTransactionDistributedTest.java
@@ -0,0 +1,379 @@
+/*
+ * 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.internal.cache.partitioned.fixed;
+
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.VM.getHostName;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.catchThrowable;
+import static org.assertj.core.api.Java6Assertions.assertThat;
+import static org.assertj.core.api.Java6Assertions.assertThatThrownBy;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.assertj.core.api.Assertions;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import org.apache.geode.DataSerializable;
+import org.apache.geode.cache.CacheTransactionManager;
+import org.apache.geode.cache.EntryOperation;
+import org.apache.geode.cache.FixedPartitionAttributes;
+import org.apache.geode.cache.FixedPartitionResolver;
+import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.TransactionDataRebalancedException;
+import org.apache.geode.cache.TransactionId;
+import org.apache.geode.cache.client.ClientRegionFactory;
+import org.apache.geode.cache.client.ClientRegionShortcut;
+import org.apache.geode.cache.client.PoolFactory;
+import org.apache.geode.cache.client.PoolManager;
+import org.apache.geode.cache.client.internal.PoolImpl;
+import org.apache.geode.cache.execute.Execution;
+import org.apache.geode.cache.execute.Function;
+import org.apache.geode.cache.execute.FunctionContext;
+import org.apache.geode.cache.execute.FunctionException;
+import org.apache.geode.cache.execute.FunctionService;
+import org.apache.geode.cache.execute.RegionFunctionContext;
+import org.apache.geode.cache.execute.ResultCollector;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.distributed.internal.ServerLocation;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.TXManagerImpl;
+import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.test.dunit.Invoke;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.ClientCacheRule;
+import org.apache.geode.test.dunit.rules.DistributedDiskDirRule;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
+
+public class FixedPartitioningWithTransactionDistributedTest implements
+ Serializable {
+
+ private String hostName;
+ private String uniqueName;
+ private String regionName;
+ private VM server1;
+ private VM server2;
+ private VM accessor;
+
+ private static final String FIXED_PARTITION_NAME = "singleBucket";
+
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule();
+
+ @Rule
+ public DistributedDiskDirRule distributedDiskDir = new DistributedDiskDirRule();
+
+ @Rule
+ public CacheRule cacheRule = new CacheRule();
+
+ @Rule
+ public ClientCacheRule clientCacheRule = new ClientCacheRule();
+
+ @Rule
+ public SerializableTestName testName = new SerializableTestName();
+
+ @Before
+ public void setup() throws Exception {
+ server1 = getVM(0);
+ server2 = getVM(1);
+ accessor = getVM(2);
+ hostName = getHostName();
+ uniqueName = getClass().getSimpleName() + "_" + testName.getMethodName();
+ regionName = uniqueName + "_region";
+ Invoke.invokeInEveryVM(() -> {
+ TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = true;
+ });
+ }
+
+ @After
+ public void tearDown() {
+ Invoke.invokeInEveryVM(() -> {
+ TXManagerImpl.ALLOW_PERSISTENT_TRANSACTIONS = false;
+ cacheRule.closeAndNullCache();
+ });
+ }
+
+ @Test
+ public void executeFunctionOnMovedPrimaryBucketFailWithTransactionDataRebalancedException() {
+ createData();
+
+ server1.invoke(() -> {
+ cacheRule.closeAndNullCache();
+ });
+
+ TransactionId transactionId = server2.invoke(() -> doFunctionTransactionAndSuspend());
+
+ server1.invoke(() -> restartPrimary());
+
+ server2.invoke(() -> {
+ assertThatThrownBy(() -> resumeFunctionTransaction(transactionId))
+ .isInstanceOf(TransactionDataRebalancedException.class);
+ });
+ }
+
+ @Test
+ public void accessorExecuteFunctionOnMovedPrimaryBucketFailWithTransactionDataRebalancedException() {
+ createData();
+ server2.invoke(() -> registerFunctions());
+
+ accessor.invoke(() -> createServerRegion(false, 1, 1, true));
+ accessor.invoke(() -> registerFunctions());
+ server1.invoke(() -> {
+ cacheRule.closeAndNullCache();
+ });
+
+ TransactionId transactionId = accessor.invoke(() -> doFunctionTransactionAndSuspend());
+
+ server1.invoke(() -> restartPrimary());
+
+ accessor.invoke(() -> {
+ assertThatThrownBy(() -> resumeFunctionTransaction(transactionId))
+ .isInstanceOf(TransactionDataRebalancedException.class);
+ });
+ }
+
+ @Test
+ public void clientExecuteFunctionOnMovedPrimaryBucketFailWithTransactionDataRebalancedException() {
+ server1.invoke(() -> createServerRegion(true, 1, 1));
+ int port2 = server2.invoke(() -> createServerRegion(false, 1, 1));
+ int port1 = accessor.invoke(() -> createServerRegion(false, 1, 1, true));
+ server1.invoke(() -> doPuts());
+ server1.invoke(() -> {
+ cacheRule.closeAndNullCache();
+ });
+ server2.invoke(() -> registerFunctions());
+ accessor.invoke(() -> registerFunctions());
+
+
+ createClientRegion(true, port1, port2);
+ Region region = clientCacheRule.getClientCache().getRegion(regionName);
+ CacheTransactionManager txManager =
+ clientCacheRule.getClientCache().getCacheTransactionManager();
+ TransactionId transactionId = doFunctionTransactionAndSuspend(region, txManager);
+
+ accessor.invoke(() -> {
+ cacheRule.closeAndNullCache();
+ });
+ server1.invoke(() -> restartPrimary());
+
+ Throwable caughtException =
+ catchThrowable(() -> resumeFunctionTransaction(transactionId, region, txManager));
+ Assertions.assertThat(caughtException).isInstanceOf(FunctionException.class);
+ Assertions.assertThat(caughtException.getCause())
+ .isInstanceOf(TransactionDataRebalancedException.class);
+ }
+
+ private void restartPrimary() throws Exception {
+ createServerRegion(true, 1, 1);
+ PartitionedRegion partitionedRegion =
+ (PartitionedRegion) cacheRule.getCache().getRegion(regionName);
+ assertThat(partitionedRegion.get(1)).isEqualTo(1);
+ await().until(() -> partitionedRegion.getBucketPrimary(0)
+ .equals(cacheRule.getCache().getInternalDistributedSystem().getDistributedMember()));
+ }
+
+ private void createData() {
+ server1.invoke(() -> createServerRegion(true, 1, 1));
+ server2.invoke(() -> createServerRegion(false, 1, 1));
+ server1.invoke(() -> doPuts());
+ }
+
+ private void doPuts() {
+ Region region = cacheRule.getCache().getRegion(regionName);
+ region.put(1, 1);
+ region.put(2, 2);
+ region.put(3, 3);
+ }
+
+ private int createServerRegion(boolean isPrimary, int redundantCopies, int totalNumOfBuckets)
+ throws Exception {
+ return createServerRegion(isPrimary, redundantCopies, totalNumOfBuckets, false);
+ }
+
+ private int createServerRegion(boolean isPrimary, int redundantCopies, int totalNumOfBuckets,
+ boolean isAccessor) throws Exception {
+ FixedPartitionAttributes fixedPartition =
+ FixedPartitionAttributes.createFixedPartition(FIXED_PARTITION_NAME, isPrimary,
+ totalNumOfBuckets);
+ PartitionAttributesFactory factory = new PartitionAttributesFactory();
+ factory.setRedundantCopies(redundantCopies).setTotalNumBuckets(totalNumOfBuckets)
+ .setPartitionResolver(new MyFixedPartitionResolver());
+ if (isAccessor) {
+ factory.setLocalMaxMemory(0);
+ } else {
+ factory.addFixedPartitionAttributes(fixedPartition);
+ }
+
+ cacheRule.getOrCreateCache().createRegionFactory(
+ isAccessor ? RegionShortcut.PARTITION : RegionShortcut.PARTITION_PERSISTENT)
+ .setPartitionAttributes(factory.create()).create(regionName);
+ CacheServer server = cacheRule.getCache().addCacheServer();
+ server.setPort(0);
+ server.start();
+ return server.getPort();
+ }
+
+ private void createClientRegion(boolean connectToFirstPort, int... ports) {
+ clientCacheRule.createClientCache();
+
+ CacheServerTestUtil.disableShufflingOfEndpoints();
+ PoolImpl pool;
+ try {
+ pool = getPool(ports);
+ } finally {
+ CacheServerTestUtil.enableShufflingOfEndpoints();
+ }
+
+ ClientRegionFactory crf =
+ clientCacheRule.getClientCache().createClientRegionFactory(ClientRegionShortcut.LOCAL);
+ crf.setPoolName(pool.getName());
+ crf.create(regionName);
+
+ if (ports.length > 1 && connectToFirstPort) {
+ // first connection to the first port in the list
+ pool.acquireConnection(new ServerLocation(hostName, ports[0]));
+ }
+ }
+
+ private PoolImpl getPool(int... ports) {
+ PoolFactory factory = PoolManager.createFactory();
+ for (int port : ports) {
+ factory.addServer(hostName, port);
+ }
+ factory.setPRSingleHopEnabled(false);
+ return (PoolImpl) factory.create(uniqueName);
+ }
+
+ private void registerFunctions() {
+ FunctionService.registerFunction(new MySuspendTransactionFunction());
+ FunctionService.registerFunction(new MyResumeTransactionFunction());
+ }
+
+ private TransactionId doFunctionTransactionAndSuspend() {
+ Region region = cacheRule.getCache().getRegion(regionName);
+ TXManagerImpl manager = cacheRule.getCache().getTxManager();
+ return doFunctionTransactionAndSuspend(region, manager);
+ }
+
+ private TransactionId doFunctionTransactionAndSuspend(Region region,
+ CacheTransactionManager manager) {
+ Execution execution = FunctionService.onRegion(region);
+ manager.begin();
+ Set keySet = new HashSet();
+ keySet.add(2);
+ ResultCollector resultCollector = execution.execute(new MySuspendTransactionFunction());
+ resultCollector.getResult();
+ return manager.suspend();
+ }
+
+ private void resumeFunctionTransaction(TransactionId transactionId) {
+ Region region = cacheRule.getCache().getRegion(regionName);
+ TXManagerImpl manager = cacheRule.getCache().getTxManager();
+ resumeFunctionTransaction(transactionId, region, manager);
+ }
+
+ private void resumeFunctionTransaction(TransactionId transactionId, Region region,
+ CacheTransactionManager manager) {
+ Execution execution = FunctionService.onRegion(region);
+ manager.resume(transactionId);
+ try {
+ Set keySet = new HashSet();
+ keySet.add(3);
+ ResultCollector resultCollector =
+ execution.withFilter(keySet).execute(new MyResumeTransactionFunction());
+ resultCollector.getResult();
+ } catch (Exception e) {
+ LogService.getLogger().info("got exception ", e);
+ throw e;
+ } finally {
+ manager.rollback();
+ }
+ }
+
+ public static class MyFixedPartitionResolver implements FixedPartitionResolver {
+
+ public MyFixedPartitionResolver() {}
+
+ @Override
+ public String getPartitionName(final EntryOperation opDetails,
+ @Deprecated final Set targetPartitions) {
+ return FIXED_PARTITION_NAME;
+ }
+
+ @Override
+ public Object getRoutingObject(final EntryOperation opDetails) {
+ return opDetails.getKey();
+ }
+
+ @Override
+ public String getName() {
+ return getClass().getName();
+ }
+ }
+
+ public static class MySuspendTransactionFunction implements Function, DataSerializable {
+ @Override
+ public void execute(FunctionContext context) {
+ assertThat(context instanceof RegionFunctionContext);
+ PartitionedRegion region = (PartitionedRegion) ((RegionFunctionContext) context).getDataSet();
+ region.containsValueForKey(2);
+ context.getResultSender().lastResult(Boolean.TRUE);
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+
+ }
+ }
+
+ public static class MyResumeTransactionFunction implements Function, DataSerializable {
+ @Override
+ public void execute(FunctionContext context) {
+ assertThat(context instanceof RegionFunctionContext);
+ PartitionedRegion region = (PartitionedRegion) ((RegionFunctionContext) context).getDataSet();
+ region.containsValueForKey(3);
+ context.getResultSender().lastResult(Boolean.TRUE);
+ }
+
+ @Override
+ public void toData(DataOutput out) throws IOException {
+
+ }
+
+ @Override
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+
+ }
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
index 6af4a31..51c6ad5 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
@@ -56,6 +56,7 @@ import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.cache.entries.AbstractRegionEntry;
+import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.Message;
import org.apache.geode.internal.concurrent.ConcurrentHashSet;
import org.apache.geode.internal.logging.LogService;
@@ -340,6 +341,9 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
proxy = new TXStateProxyImpl(cache, this, id, null);
}
setTXState(proxy);
+ if (logger.isDebugEnabled()) {
+ logger.debug("begin tx: " + proxy);
+ }
this.localTxMap.put(id, proxy);
}
@@ -879,7 +883,9 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
}
}
-
+ if (logger.isDebugEnabled()) {
+ logger.debug("masqueradeAs tx {} for msg {} ", val, msg);
+ }
setTXState(val);
return val;
}
@@ -896,6 +902,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
} else {
val = new TXStateProxyImpl(cache, this, key, msg.getTXOriginatorClient());
val.setLocalTXState(new TXState(val, true));
+ val.setTarget(cache.getDistributedSystem().getDistributedMember());
}
this.hostedTXStates.put(key, val);
}
@@ -980,6 +987,10 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
setTXState(val);
}
+ if (logger.isDebugEnabled()) {
+ logger.debug("masqueradeAs tx {} for client message {}", val,
+ MessageType.getString(msg.getMessageType()));
+ }
return val;
}
@@ -994,6 +1005,9 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
txState.getLock().lock();
}
setTXState(txState);
+ if (logger.isDebugEnabled()) {
+ logger.debug("masqueradeAs tx " + txState);
+ }
}
/**
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
index 7956416..645ba36 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
@@ -140,7 +140,6 @@ public class TXStateProxyImpl implements TXStateProxy {
// wait for the region to be initialized fixes bug 44652
r.waitOnInitialization(r.getInitializationLatchBeforeGetInitialImage());
target = r.getOwnerForKey(key);
-
if (target == null || target.equals(this.txMgr.getDM().getId())) {
this.realDeal = new TXState(this, false);
} else {
@@ -724,6 +723,10 @@ public class TXStateProxyImpl implements TXStateProxy {
public void setTarget(DistributedMember target) {
assert this.target == null;
getRealDeal(target);
+ if (this.target == null && isRealDealLocal()) {
+ assert target.equals(getCache().getDistributedSystem().getDistributedMember());
+ this.target = target;
+ }
}
@Override
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/TXManagerImplTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/TXManagerImplTest.java
index 9b56dfd..b198ce1 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/TXManagerImplTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/TXManagerImplTest.java
@@ -94,6 +94,7 @@ public class TXManagerImplTest {
InternalDistributedSystem distributedSystem = mock(InternalDistributedSystem.class);
doReturn(distributedSystem).when(spyCache).getDistributedSystem();
when(distributedSystem.getDistributionManager()).thenReturn(dm);
+ when(distributedSystem.getDistributedMember()).thenReturn(member);
spyTxMgr = spy(new TXManagerImpl(mock(CachePerfStats.class), spyCache));
timer = mock(SystemTimer.class);
doReturn(timer).when(spyCache).getCCPTimer();
@@ -518,4 +519,12 @@ public class TXManagerImplTest {
verify(lock, times(1)).unlock();
}
+
+ @Test
+ public void masqueradeAsSetsTarget() throws InterruptedException {
+ TXStateProxy tx;
+
+ tx = txMgr.masqueradeAs(msg);
+ assertNotNull(tx.getTarget());
+ }
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplTest.java
index 547f5ce..b55e745 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplTest.java
@@ -23,6 +23,9 @@ import static org.mockito.Mockito.when;
import org.junit.Before;
import org.junit.Test;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.DistributedSystem;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.LocalRegion.NonTXEntry;
import org.apache.geode.internal.cache.region.entry.RegionEntryFactoryBuilder;
@@ -36,6 +39,8 @@ public class TXStateProxyImplTest {
LocalRegionDataView view;
private TXId txId;
private TXManagerImpl txManager;
+ private DistributedSystem system;
+ private DistributedMember member;
@Before
public void setUp() {
@@ -44,6 +49,11 @@ public class TXStateProxyImplTest {
txId = new TXId(mock(InternalDistributedMember.class), 1);
txManager = mock(TXManagerImpl.class);
view = mock(LocalRegionDataView.class);
+ system = mock(InternalDistributedSystem.class);
+ member = mock(InternalDistributedMember.class);
+
+ when(cache.getDistributedSystem()).thenReturn(system);
+ when(system.getDistributedMember()).thenReturn(member);
}
@Test
@@ -99,4 +109,53 @@ public class TXStateProxyImplTest {
assertThat(tx.isOverTransactionTimeoutLimit()).isEqualTo(false);
}
+
+ @Test
+ public void setTargetWillSetTargetToItselfAndSetTXStateIfRealDealIsNull() {
+ TXStateProxyImpl tx = spy(new TXStateProxyImpl(cache, txManager, txId, false));
+ assertThat(tx.hasRealDeal()).isFalse();
+ assertThat(tx.getTarget()).isNull();
+
+ tx.setTarget(member);
+
+ assertThat(tx.getTarget()).isEqualTo(member);
+ assertThat(tx.isRealDealLocal()).isTrue();
+ }
+
+ @Test
+ public void setTargetWillSetTXStateStubIfTargetIsDifferentFromLocalMember() {
+ TXStateProxyImpl tx = spy(new TXStateProxyImpl(cache, txManager, txId, false));
+ assertThat(tx.hasRealDeal()).isFalse();
+ assertThat(tx.getTarget()).isNull();
+ DistributedMember remoteMember = mock(InternalDistributedMember.class);
+
+ tx.setTarget(remoteMember);
+
+ assertThat(tx.getTarget()).isEqualTo(remoteMember);
+ assertThat(tx.isRealDealLocal()).isFalse();
+ assertThat(tx.hasRealDeal()).isTrue();
+ }
+
+ @Test
+ public void setTargetToItSelfIfRealDealIsTXStateAndTargetIsSameAsLocalMember() {
+ TXStateProxyImpl tx = spy(new TXStateProxyImpl(cache, txManager, txId, false));
+ tx.setLocalTXState(new TXState(tx, true));
+ assertThat(tx.isRealDealLocal()).isTrue();
+ assertThat(tx.getTarget()).isNull();
+
+ tx.setTarget(member);
+
+ assertThat(tx.getTarget()).isEqualTo(member);
+ assertThat(tx.isRealDealLocal()).isTrue();
+ }
+
+ @Test(expected = AssertionError.class)
+ public void setTargetThrowsIfIfRealDealIsTXStateAndTargetIsDifferentFromLocalMember() {
+ TXStateProxyImpl tx = spy(new TXStateProxyImpl(cache, txManager, txId, false));
+ tx.setLocalTXState(new TXState(tx, true));
+ assertThat(tx.getTarget()).isNull();
+ DistributedMember remoteMember = mock(InternalDistributedMember.class);
+
+ tx.setTarget(remoteMember);
+ }
}
diff --git a/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/CacheRule.java b/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/CacheRule.java
index 7b6ce63..4169932 100644
--- a/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/CacheRule.java
+++ b/geode-dunit/src/main/java/org/apache/geode/test/dunit/rules/CacheRule.java
@@ -210,7 +210,7 @@ public class CacheRule extends AbstractDistributedRule {
return cache;
}
- private void closeAndNullCache() {
+ public void closeAndNullCache() {
closeCache();
nullCache();
}