@Test
  public void testMoveBucketsOverflowToDisk() {
    VM vm0 = getVM(0);
    VM vm1 = getVM(1);
    VM vm2 = getVM(2);
    VM vm3 = getVM(3);

    // Create the region in two VMs
    vm0.invoke(
        () -> createPartitionedRegion("region1", createLRUEntryAttributes(1, OVERFLOW_TO_DISK)));
    vm1.invoke(
        () -> createPartitionedRegion("region1", createLRUEntryAttributes(1, OVERFLOW_TO_DISK)));

    // Create some buckets
    vm0.invoke(() -> {
      Region<Number, String> region = getCache().getRegion("region1");
      for (int bucket = 0; bucket < 12; bucket++) {
        Map<Number, String> map = new HashMap<>();
        for (int key = 0; key < 200; key++) {
          map.put(bucket + 113 * key, "A");
        }
        region.putAll(map);
      }
    });

    // Do some puts and gets, to trigger eviction

    // Do some operations
    vm0.invoke(() -> {
      Region<Number, String> region = getCache().getRegion("region1");

      Random random = new Random();

      for (int count = 0; count < 5000; count++) {
        int bucket = count % 12;
        int key = random.nextInt(20);
        region.put(bucket + 113 * key, "B");
      }

      for (int count = 0; count < 500; count++) {
        int bucket = count % 12;
        int key = random.nextInt(20);
        region.get(bucket + 113 * key);
      }
    });

    // Create the region in one more VM.
    vm2.invoke(
        () -> createPartitionedRegion("region1", createLRUEntryAttributes(1, OVERFLOW_TO_DISK)));

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

      assertThat(results.getTotalBucketCreatesCompleted()).isEqualTo(0);

      // We don't know how many primaries will move, it depends on
      // if the move bucket code moves the primary or a redundant bucket
      // assertIndexDetailsEquals(0, results.getTotalPrimaryTransfersCompleted());
      assertThat(results.getTotalBucketTransfersCompleted()).isEqualTo(8);
      assertThat(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.getBucketTransferBytes()).isGreaterThan(0);
      assertThat(details.getBucketTransfersCompleted()).isEqualTo(8);

      long value = 0;
      Set<PartitionMemberInfo> beforeDetails = details.getPartitionMemberDetailsAfter();
      for (PartitionMemberInfo memberDetails : beforeDetails) {
        value += memberDetails.getSize();
      }

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

      long afterSize = 0;
      for (PartitionMemberInfo memberDetails : afterDetails) {
        assertThat(memberDetails.getBucketCount()).isEqualTo(8);
        assertThat(memberDetails.getPrimaryCount()).isEqualTo(4);
        afterSize += memberDetails.getSize();
      }
      assertThat(afterSize).isEqualTo(value);

      validateStatistics(manager, results);
    });

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

        PartitionRegionInfo details = PartitionRegionHelper.getPartitionRegionInfo(region);
        assertThat(details.getCreatedBucketCount()).isEqualTo(12);
        assertThat(details.getActualRedundantCopies()).isEqualTo(1);
        assertThat(details.getLowRedundancyBucketCount()).isEqualTo(0);

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

    // Create the region in one more VM.
    vm3.invoke(
        () -> createPartitionedRegion("region1", createLRUEntryAttributes(1, OVERFLOW_TO_DISK)));

    // Do another rebalance
    vm0.invoke(() -> {
      ResourceManager manager = getCache().getResourceManager();
      RebalanceResults results = doRebalance(false, manager);

      assertThat(results.getTotalBucketCreatesCompleted()).isEqualTo(0);

      // We don't know how many primaries will move, it depends on
      // if the move bucket code moves the primary or a redundant bucket
      // assertIndexDetailsEquals(0, results.getTotalPrimaryTransfersCompleted());
      assertThat(results.getTotalBucketTransfersCompleted()).isEqualTo(6);
      assertThat(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.getBucketTransferBytes()).isGreaterThan(0);
      assertThat(details.getBucketTransfersCompleted()).isEqualTo(6);

      long totalSize = 0;
      Set<PartitionMemberInfo> beforeDetails = details.getPartitionMemberDetailsAfter();
      for (PartitionMemberInfo memberDetails : beforeDetails) {
        totalSize += memberDetails.getSize();
      }

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

      long afterSize = 0;
      for (PartitionMemberInfo memberDetails : afterDetails) {
        assertThat(memberDetails.getBucketCount()).isEqualTo(6);

        afterSize += memberDetails.getSize();
      }
      assertThat(afterSize).isEqualTo(totalSize);
    });

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

        PartitionRegionInfo details = PartitionRegionHelper.getPartitionRegionInfo(region);
        assertThat(details.getCreatedBucketCount()).isEqualTo(12);
        assertThat(details.getActualRedundantCopies()).isEqualTo(1);
        assertThat(details.getLowRedundancyBucketCount()).isEqualTo(0);

        for (PartitionMemberInfo memberDetails : details.getPartitionMemberInfo()) {
          assertThat(memberDetails.getBucketCount()).isEqualTo(6);
        }
      });
    }
  }
Verifies when overflowing entries to disk that the stats are correct and we still rebalance
correctly
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java