FetchKeysMessage extends PartitionMessage { public static FetchKeysResponse send(InternalDistributedMember recipient, PartitionedRegion r, Integer bucketId, boolean allowTombstones) throws ForceReattemptException { Assert.assertTrue(recipient != null, "FetchKeysMessage NULL recipient"); TXManagerImpl txManager = r.getCache().getTxManager(); boolean resetTxState = isTransactionInternalSuspendNeeded(txManager); TXStateProxy txStateProxy = null; if (resetTxState) { txStateProxy = txManager.pauseTransaction(); } try { FetchKeysMessage tmp = new FetchKeysMessage(); FetchKeysResponse p = (FetchKeysResponse) tmp.createReplyProcessor(r, Collections.singleton(recipient)); FetchKeysMessage m = new FetchKeysMessage(recipient, r.getPRId(), p, bucketId, InterestType.REGULAR_EXPRESSION, ".*", allowTombstones); m.setTransactionDistributed(txManager.isDistributed()); Set failures = r.getDistributionManager().putOutgoing(m); if (failures != null && failures.size() > 0) { throw new ForceReattemptException( String.format("Failed sending < %s >", m)); } return p; } finally { if (resetTxState) { txManager.unpauseTransaction(txStateProxy); } } } private  FetchKeysMessage(InternalDistributedMember recipient, int regionId,
      ReplyProcessor21 processor, Integer bucketId, int itype, Object interestArg,
      boolean allowTombstones);  FetchKeysMessage(); static FetchKeysResponse send(InternalDistributedMember recipient, PartitionedRegion r,
      Integer bucketId, boolean allowTombstones); static FetchKeysResponse sendInterestQuery(InternalDistributedMember recipient,
      PartitionedRegion r, Integer bucketId, int itype, Object arg, boolean allowTombstones); @Override int getDSFID(); @Override KnownVersion[] getSerializationVersions(); @Override void fromData(DataInput in,
      DeserializationContext context); @Override void toData(DataOutput out,
      SerializationContext context); public KnownVersion[] serializationVersions; }
@Test public void sendsWithTransactionPaused_ifTransactionIsHostedLocally() throws Exception { when(txStateProxy.isRealDealLocal()).thenReturn(true); when(txStateProxy.isDistTx()).thenReturn(false); FetchKeysMessage.send(recipient, region, 1, false); InOrder inOrder = inOrder(txManager, distributionManager); inOrder.verify(txManager, times(1)).pauseTransaction(); inOrder.verify(distributionManager, times(1)).putOutgoing(sentMessage.capture()); inOrder.verify(txManager, times(1)).unpauseTransaction(same(txStateProxy)); assertThat(sentMessage.getValue().getTXUniqId()).isEqualTo(TXManagerImpl.NOTX); }
@Test public void sendsWithoutPausingTransaction_ifTransactionIsNotHostedLocally() throws Exception { when(txStateProxy.isRealDealLocal()).thenReturn(false); int uniqueId = 99; TXId txID = new TXId(recipient, uniqueId); when(txStateProxy.getTxId()).thenReturn(txID); FetchKeysMessage.send(recipient, region, 1, false); verify(distributionManager, times(1)).putOutgoing(sentMessage.capture()); assertThat(sentMessage.getValue().getTXUniqId()).isEqualTo(uniqueId); verify(txManager, never()).pauseTransaction(); verify(txManager, never()).unpauseTransaction(any()); }