private void doOpDuringBucketRemove(final OpDuringBucketRemove op) {
    final String[] regionPath = new String[] {getUniqueName() + "_PR_0"};
    final int[] numBuckets = new int[] {1};
    final int[] redundantCopies = new int[] {1};

    // localMaxMemory config to use for three members
    final int[] localMaxMemory = new int[] {100, 100};

    final Integer KEY = 69;
    // bucketKeys to use for making one bucket
    final Integer[] bucketKeys = new Integer[] {KEY};

    createRegion(Host.getHost(0).getVM(0), regionPath[0], localMaxMemory[0], numBuckets[0],
        redundantCopies[0]);
    createRegion(Host.getHost(0).getVM(1), regionPath[0], localMaxMemory[1], numBuckets[0],
        redundantCopies[0]);

    final String VALUE = "doOpDuringBucketRemove.VALUE";

    createBuckets(0, regionPath[0], bucketKeys, VALUE);

    // identify the members and their config values
    final InternalDistributedMember[] members = new InternalDistributedMember[2];
    final long[] memberSizes = new long[members.length];
    final int[] memberBucketCounts = new int[members.length];
    final int[] memberPrimaryCounts = new int[members.length];

    fillValidationArrays(members, memberSizes, memberBucketCounts, memberPrimaryCounts,
        regionPath[0]);

    int primaryVM = -1;
    int otherVM = -1;
    for (int i = 0; i < memberPrimaryCounts.length; i++) {
      if (memberPrimaryCounts[i] == 0) {
        otherVM = i;
      } else if (memberPrimaryCounts[i] == 1) {
        // found the primary
        primaryVM = i;
      }
    }

    assertTrue(primaryVM > -1);
    assertTrue(otherVM > -1);
    assertTrue(primaryVM != otherVM);

    final int finalOtherVM = otherVM;

    Host.getHost(0).getVM(otherVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(regionPath[0]);
        Bucket bucket = pr.getRegionAdvisor().getBucket(0);
        assertTrue("Target member is not hosting bucket to remove", bucket.isHosting());
        assertNotNull("Bucket is null on target member", bucket);
        assertNotNull("BucketRegion is null on target member",
            bucket.getBucketAdvisor().getProxyBucketRegion().getHostedBucketRegion());
        EntrySnapshot re = (EntrySnapshot) pr.getEntry(KEY);
        assertEquals(true, re.wasInitiallyLocal());
        assertEquals(false, re.isLocal());
        assertEquals(VALUE, re.getValue());
      }
    });
    Host.getHost(0).getVM(otherVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        final PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(regionPath[0]);
        final boolean[] invoked = new boolean[] {false};
        final PartitionedRegionDataStore prds = pr.getDataStore();
        prds.setBucketReadHook(() -> {
          invoked[0] = true;
          logger.debug("In bucketReadHook");
          assertTrue(prds.removeBucket(0, false));
        });
        try {
          Bucket bucket = pr.getRegionAdvisor().getBucket(0);
          assertTrue("Target member is not hosting bucket to remove", bucket.isHosting());
          assertNotNull("Bucket is null on target member", bucket);
          assertNotNull("BucketRegion is null on target member",
              bucket.getBucketAdvisor().getProxyBucketRegion().getHostedBucketRegion());
          assertEquals(false, invoked[0]);
          op.runit(pr, KEY, VALUE);
          assertEquals(true, invoked[0]);
        } finally {
          prds.setBucketReadHook(null);
        }
      }
    });

    Host.getHost(0).getVM(otherVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(regionPath[0]);

        Bucket bucket = pr.getRegionAdvisor().getBucket(0);
        BucketRegion bucketRegion =
            bucket.getBucketAdvisor().getProxyBucketRegion().getHostedBucketRegion();

        assertFalse("Target member is still hosting removed bucket. Bucket:" + bucket
            + " Advisor state:" + bucket.getBucketAdvisor(), bucket.isHosting());

        assertNull(bucketRegion);
        // assertTrue(bucketRegion == null || bucketRegion.isDestroyed());
        EntrySnapshot re = (EntrySnapshot) pr.getEntry(KEY);
        assertEquals(false, re.wasInitiallyLocal());
        assertEquals(false, re.isLocal());
        assertEquals(VALUE, re.getValue());
      }
    });
  }
Creates partitioned regions in multiple vms and fully exercises the internal-only
getInternalPRDetails API on ResourceManager.
geode-core/src/distributedTest/java/org/apache/geode/cache/management/ResourceManagerDUnitTest.java