@Test
  @Parameters({"true", "false"})
  @TestCaseName("{method}(simulate={0})")
  public void testMoveBucketsWithUnrecoveredValues(boolean simulate) {
    VM vm0 = getVM(0);
    VM vm1 = getVM(1);

    // Create the region in only 1 VM
    vm0.invoke(() -> createPersistentPartitionedRegion("region1", "store", getDiskDirs(),
        new NullReturningLoader<>()));

    // Create some buckets
    vm0.invoke(() -> {
      Region<Number, String> region = getCache().getRegion("region1");
      region.put(1, "A");
      region.put(2, "A");
      region.put(3, "A");
      region.put(4, "A");
      region.put(5, "A");
      region.put(6, "A");
    });

    vm0.invoke(() -> {
      PartitionedRegion region = (PartitionedRegion) getCache().getRegion("region1");
      PartitionedRegionDataStore dataStore = region.getDataStore();

      for (int i = 1; i <= 6; i++) {
        BucketRegion bucket = dataStore.getLocalBucketById(i);

        assertThat(bucket.getNumOverflowBytesOnDisk()).isEqualTo(0);
        assertThat(bucket.getNumOverflowOnDisk()).isEqualTo(0);
        assertThat(bucket.getNumEntriesInVM()).isEqualTo(1);
      }
      getCache().close();
    });

    // now recover the region
    vm0.invoke(() -> createPersistentPartitionedRegion("region1", "store", getDiskDirs(),
        new NullReturningLoader<>()));

    vm0.invoke(() -> {
      PartitionedRegion region = (PartitionedRegion) getCache().getRegion("region1");
      PartitionedRegionDataStore dataStore = region.getDataStore();

      for (int i = 1; i <= 6; i++) {
        BucketRegion bucket = dataStore.getLocalBucketById(i);

        assertThat(bucket.getNumOverflowOnDisk()).isEqualTo(1);
        assertThat(bucket.getNumEntriesInVM()).isEqualTo(0);

        // the size recorded on disk is not the same is the in memory size, apparently
        assertThat(bucket.getNumOverflowBytesOnDisk())
            .as("Bucket size was " + bucket.getNumOverflowBytesOnDisk())
            .isGreaterThan(1);
        assertThat(bucket.getTotalBytes()).isEqualTo(bucket.getNumOverflowBytesOnDisk());
      }
    });

    // Create the region in the other VM (should have no effect)
    vm1.invoke(() -> createPersistentPartitionedRegion("region1", "store", getDiskDirs(),
        new NullReturningLoader<>()));

    // Now simulate a rebalance
    vm0.invoke(() -> {
      InternalResourceManager manager = getCache().getInternalResourceManager();
      RebalanceResults results = doRebalance(simulate, manager);

      assertThat(results.getTotalBucketCreatesCompleted()).isEqualTo(0);
      assertThat(results.getTotalPrimaryTransfersCompleted()).isEqualTo(0);
      assertThat(results.getTotalBucketTransfersCompleted()).isEqualTo(3);
      assertThat(results.getTotalBucketTransferBytes())
          .as("Transferred Bytes = " + results.getTotalBucketTransferBytes())
          .isGreaterThan(0);

      Set<PartitionRebalanceInfo> detailSet = results.getPartitionRebalanceDetails();
      assertThat(detailSet).hasSize(1);

      PartitionRebalanceInfo details = detailSet.iterator().next();
      assertThat(details.getBucketCreatesCompleted()).isEqualTo(0);
      assertThat(details.getPrimaryTransfersCompleted()).isEqualTo(0);
      assertThat(details.getBucketTransferBytes()).isGreaterThan(0);
      assertThat(details.getBucketTransfersCompleted()).isEqualTo(3);

      Set<PartitionMemberInfo> afterDetails = details.getPartitionMemberDetailsAfter();
      assertThat(afterDetails).hasSize(2);

      for (PartitionMemberInfo memberDetails : afterDetails) {
        assertThat(memberDetails.getBucketCount()).isEqualTo(3);
        assertThat(memberDetails.getPrimaryCount()).isEqualTo(3);
      }

      if (!simulate) {
        validateStatistics(manager, results);
      }
    });

    if (!simulate) {
      for (VM vm : toArray(vm0, vm1)) {
        vm.invoke(() -> {
          Region region = getCache().getRegion("region1");

          PartitionRegionInfo details = PartitionRegionHelper.getPartitionRegionInfo(region);
          assertThat(details.getCreatedBucketCount()).isEqualTo(6);
          assertThat(details.getActualRedundantCopies()).isEqualTo(0);
          assertThat(details.getLowRedundancyBucketCount()).isEqualTo(0);
          assertThat(details.getPartitionMemberInfo()).hasSize(2);

          for (PartitionMemberInfo memberDetails : details.getPartitionMemberInfo()) {
            assertThat(memberDetails.getBucketCount()).isEqualTo(3);
            assertThat(memberDetails.getPrimaryCount()).isEqualTo(3);
          }

          // check to make sure that moving buckets didn't close the cache loader
          NullReturningLoader loader =
              (NullReturningLoader) getCache().getRegion("region1").getAttributes()
                  .getCacheLoader();
          assertThat(loader.isClosed()).isFalse();
        });
      }

      for (VM vm : toArray(vm0, vm1)) {
        vm.invoke(() -> {
          NullReturningLoader loader =
              (NullReturningLoader) getCache().getRegion("region1").getAttributes()
                  .getCacheLoader();
          assertThat(loader.isClosed()).isFalse();

          // check to make sure that closing the PR closes the cache loader
          getCache().getRegion("region1").close();
          assertThat(loader.isClosed()).isTrue();
        });
      }
    }
  }
Test to ensure that we wait for in progress write operations before moving a primary.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java