@Override
      public Object call() {
        // do local operations with rollback and then commit
        Map<CustId, Customer> custMap = new HashMap<>();
        Region<CustId, Customer> custRegion = getCache().getRegion(CUSTOMER);
        PartitionedRegion pr = ((PartitionedRegion) custRegion);
        List localBuckets = pr.getLocalPrimaryBucketsListTestOnly();
        System.out.println("localBuckets:" + localBuckets);
        for (int i = 10; i < 20; i++) {
          int hash = PartitionedRegionHelper.getHashKey(i,
              pr.getPartitionAttributes().getTotalNumBuckets());
          if (localBuckets.contains(hash)) {
            custMap.put(new CustId(i), new Customer("name" + i, "address" + i));
          }
        }
        System.out.println("SWAP:custMap:" + custMap);
        int regionSize = custRegion.size();
        getCache().getCacheTransactionManager().begin();
        if (op.equals(OP.PUTALL)) {
          custRegion.putAll(custMap);
        } else {
          custRegion.removeAll(custMap.keySet());
        }
        getCache().getCacheTransactionManager().rollback();
        assertThat(regionSize).isEqualTo(custRegion.size());
        // now commit
        getCache().getCacheTransactionManager().begin();
        if (op.equals(OP.PUTALL)) {
          custRegion.putAll(custMap);
        } else {
          custRegion.removeAll(custMap.keySet());
        }
        getCache().getCacheTransactionManager().commit();
        assertThat(getExpectedSize(custMap, regionSize)).isEqualTo(custRegion.size());

        // bulk op on other member
        custMap.clear();
        for (int i = 10; i < 20; i++) {
          int hash = PartitionedRegionHelper.getHashKey(i,
              pr.getPartitionAttributes().getTotalNumBuckets());
          if (!localBuckets.contains(hash)) { // not on local member
            custMap.put(new CustId(i), new Customer("name" + i, "address" + i));
          }
        }
        System.out.println("SWAP:custMap:" + custMap);
        regionSize = custRegion.size();
        getCache().getCacheTransactionManager().begin();
        if (op.equals(OP.PUTALL)) {
          custRegion.putAll(custMap);
        } else {
          custRegion.removeAll(custMap.keySet());
        }
        getCache().getCacheTransactionManager().rollback();
        assertThat(regionSize).isEqualTo(custRegion.size());
        // now commit
        getCache().getCacheTransactionManager().begin();
        if (op.equals(OP.PUTALL)) {
          custRegion.putAll(custMap);
        } else {
          custRegion.removeAll(custMap.keySet());
        }
        getCache().getCacheTransactionManager().commit();
        assertThat(getExpectedSize(custMap, regionSize)).isEqualTo(custRegion.size());
        return localBuckets;
      }
disabled because rather than throwing an exception, getAll catches all exceptions and logs a
warning message
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/RemoteTransactionDUnitTest.java