@Test
  public void testTxFunctionWithOtherOps() {
    Host host = Host.getHost(0);
    VM accessor = host.getVM(0);
    VM datastore1 = host.getVM(1);
    VM datastore2 = host.getVM(2);
    initAccessorAndDataStore(accessor, datastore1, datastore2, 0);

    SerializableCallable<Object> registerFunction = new SerializableCallable<Object>() {
      @Override
      public Object call() {
        FunctionService.registerFunction(new TXFunction());
        return null;
      }
    };

    accessor.invoke(registerFunction);
    datastore1.invoke(registerFunction);
    datastore2.invoke(registerFunction);

    TXId txId = (TXId) accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        Region custRegion = getCache().getRegion(CUSTOMER);
        TXManagerImpl mgr = getCache().getTXMgr();
        mgr.begin();
        assertThatThrownBy(
            () -> FunctionService.onRegion(custRegion).execute(TXFunction.id).getResult())
                .isInstanceOf(TransactionException.class);
        Set filter = new HashSet();
        filter.add(expectedCustId);
        FunctionService.onRegion(custRegion).withFilter(filter).execute(TXFunction.id).getResult();
        assertThat(expectedCustomer).isEqualTo(custRegion.get(expectedCustId));
        TXStateProxy tx = mgr.pauseTransaction();
        assertThat(custRegion.get(expectedCustId)).isNull();
        mgr.unpauseTransaction(tx);
        return mgr.suspend();
      }
    });

    final Integer txOnDatastore1 = (Integer) datastore1.invoke(getNumberOfTXInProgress);
    final Integer txOnDatastore2 = (Integer) datastore2.invoke(getNumberOfTXInProgress);
    assertThat(txOnDatastore1 + txOnDatastore2).isOne();

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        Region custRegion = getCache().getRegion(CUSTOMER);
        CacheTransactionManager mgr = getCache().getTXMgr();
        mgr.resume(txId);
        mgr.commit();
        assertThat(expectedCustomer).isEqualTo(custRegion.get(expectedCustId));
        custRegion.destroy(expectedCustId);
        return null;
      }
    });
    // test onMembers
    SerializableCallable<Object> getMember = new SerializableCallable<Object>() {
      @Override
      public Object call() {
        return getCache().getMyId();
      }
    };
    final InternalDistributedMember ds1 = (InternalDistributedMember) datastore1.invoke(getMember);
    final InternalDistributedMember ds2 = (InternalDistributedMember) datastore2.invoke(getMember);
    TXId txId2 = (TXId) accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(CUSTOMER);
        // get owner for expectedKey
        DistributedMember owner = pr.getOwnerForKey(pr.getKeyInfo(expectedCustId));
        // get key on datastore1
        CustId keyOnOwner = null;
        keyOnOwner = getKeyOnMember(owner, pr);

        TXManagerImpl mgr = getCache().getTXMgr();
        mgr.begin();
        // bootstrap tx on owner
        pr.get(keyOnOwner);
        Set<DistributedMember> members = new HashSet<>();
        members.add(ds1);
        members.add(ds2);
        assertThatThrownBy(
            () -> FunctionService.onMembers(members).execute(TXFunction.id).getResult())
                .isInstanceOf(TransactionException.class);
        FunctionService.onMember(owner).execute(TXFunction.id).getResult();
        assertThat(expectedCustomer).isEqualTo(pr.get(expectedCustId));
        TXStateProxy tx = mgr.pauseTransaction();
        assertThat(pr.get(expectedCustId)).isNull();
        mgr.unpauseTransaction(tx);
        return mgr.suspend();
      }
    });
    final Integer txOnDatastore1_1 = (Integer) datastore1.invoke(getNumberOfTXInProgress);
    final Integer txOnDatastore2_1 = (Integer) datastore2.invoke(getNumberOfTXInProgress);
    assertThat(txOnDatastore1_1 + txOnDatastore2_1).isOne();

    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        Region custRegion = getCache().getRegion(CUSTOMER);
        CacheTransactionManager mgr = getCache().getTXMgr();
        mgr.resume(txId2);
        mgr.commit();
        assertThat(expectedCustomer).isEqualTo(custRegion.get(expectedCustId));
        custRegion.destroy(expectedCustId);
        return null;
      }
    });
    // test function execution on data store
    final DistributedMember owner =
        (DistributedMember) accessor.invoke(new SerializableCallable<Object>() {
          @Override
          public Object call() {
            PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(CUSTOMER);
            return pr.getOwnerForKey(pr.getKeyInfo(expectedCustId));
          }
        });

    SerializableCallable<Object> testFnOnDs = new SerializableCallable<Object>() {
      @Override
      public Object call() {
        TXManagerImpl mgr = getCache().getTXMgr();
        PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(CUSTOMER);
        CustId keyOnDs = getKeyOnMember(pr.getMyId(), pr);
        mgr.begin();
        pr.get(keyOnDs);
        Set filter = new HashSet();
        filter.add(keyOnDs);
        FunctionService.onRegion(pr).withFilter(filter).execute(TXFunction.id).getResult();
        assertThat(expectedCustomer).isEqualTo(pr.get(expectedCustId));
        TXStateProxy tx = mgr.pauseTransaction();
        assertThat(pr.get(expectedCustId)).isNull();
        mgr.unpauseTransaction(tx);
        return mgr.suspend();
      }
    };

    if (owner.equals(ds1)) {
      TXId dsTxId = (TXId) datastore1.invoke(testFnOnDs);

      final Integer txOnDatastore1_2 = (Integer) datastore1.invoke(getNumberOfTXInProgress);
      final Integer txOnDatastore2_2 = (Integer) datastore2.invoke(getNumberOfTXInProgress);
      assertThat(txOnDatastore1_2 + txOnDatastore2_2).isZero();// ds1 has a local transaction,
                                                               // not remote
      CloseFnOnDsTx closeFnOnDsTx = new CloseFnOnDsTx(dsTxId);
      datastore1.invoke(closeFnOnDsTx);
    } else {
      TXId dsTxId = (TXId) datastore2.invoke(testFnOnDs);

      final Integer txOnDatastore1_2 = (Integer) datastore1.invoke(getNumberOfTXInProgress);
      final Integer txOnDatastore2_2 = (Integer) datastore2.invoke(getNumberOfTXInProgress);
      assertThat(txOnDatastore1_2 + txOnDatastore2_2).isZero();// ds1 has a local transaction,
                                                               // not remote
      CloseFnOnDsTx closeFnOnDsTx = new CloseFnOnDsTx(dsTxId);
      datastore2.invoke(closeFnOnDsTx);
    }

    // test that function is rejected if function target is not same as txState target
    accessor.invoke(new SerializableCallable<Object>() {
      @Override
      public Object call() {
        CacheTransactionManager mgr = getCache().getTXMgr();
        PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(CUSTOMER);
        CustId keyOnDs1 = getKeyOnMember(ds1, pr);
        CustId keyOnDs2 = getKeyOnMember(ds2, pr);
        mgr.begin();
        pr.get(keyOnDs1);// bootstrap txState
        Set filter = new HashSet();
        filter.add(keyOnDs2);
        assertThatThrownBy(() -> FunctionService.onRegion(pr).withFilter(filter)
            .execute(TXFunction.id).getResult())
                .isInstanceOf(TransactionDataRebalancedException.class);
        assertThatThrownBy(() -> FunctionService.onMember(ds2).execute(TXFunction.id).getResult())
            .isInstanceOf(TransactionDataNotColocatedException.class);
        mgr.commit();
        return null;
      }
    });
  }
When we have narrowed down on a target node for a transaction, test that we throw an exception
if that node does not host primary for subsequent entries
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/RemoteTransactionDUnitTest.java