@Test
  @Parameters({"true,false", "false,false", "true,true", "false,true"})
  @TestCaseName("{method}(simulate={0}, userAccessor={1})")
  public void testRecoverRedundancyWithOfflinePersistence(boolean simulate, boolean useAccessor)
      throws Exception {
    VM vm0 = getVM(0);
    VM vm1 = getVM(1);
    VM vm2 = getVM(2);
    VM vm3 = getVM(3);

    int redundantCopies = 1;

    // Create the region in only 2 VMs
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(
          () -> createPersistentPartitionedRegion("region1", getUniqueName(), getDiskDirs(),
              redundantCopies));
    }

    VM rebalanceVM = useAccessor ? vm3 : vm0;
    if (useAccessor) {
      // Create an accessor
      rebalanceVM.invoke(() -> createAccessor("region1", "ds1"));
    }

    // 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");
    });

    // Close the cache in vm1
    Set<Integer> bucketsOnVM1 = vm1.invoke(() -> getBucketList("region1"));
    vm1.invoke(() -> getCache().getRegion("region1").close());

    // make sure we can tell that the buckets have low redundancy
    vm0.invoke(() -> validateRedundancy("region1", 6, 0, 6));

    // Now create the cache in another member
    vm2.invoke(
        () -> createPersistentPartitionedRegion("region1", getUniqueName(), getDiskDirs(),
            redundantCopies));

    // Make sure we still have low redundancy
    vm0.invoke(() -> validateRedundancy("region1", 6, 0, 6));

    /*
     * Simulates a rebalance if simulation flag is set. Otherwise, performs a rebalance.
     *
     * A rebalance will replace offline buckets, so this should restore redundancy
     */
    rebalanceVM.invoke(() -> {
      InternalResourceManager manager = getCache().getInternalResourceManager();
      RebalanceResults results = doRebalance(simulate, manager);

      assertThat(results.getTotalBucketCreatesCompleted()).isEqualTo(6);
      assertThat(results.getTotalPrimaryTransfersCompleted()).isEqualTo(3);
      assertThat(results.getTotalBucketTransfersCompleted()).isEqualTo(0);

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

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

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

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

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

    Set<Integer> bucketsOnVM0 = vm0.invoke(() -> getBucketList("region1"));
    Set<Integer> bucketsOnVM2 = vm2.invoke(() -> getBucketList("region1"));

    if (simulate) {
      // Otherwise, we should still have broken redundancy at this point
      vm0.invoke(() -> validateRedundancy("region1", 6, 0, 6));
    } else {
      // Make sure redundancy is repaired if not simulated
      vm0.invoke(() -> validateRedundancy("region1", 6, 1, 0));
    }

    vm2.invoke(() -> getCache().getRegion("region1").close());
    vm0.invoke(() -> getCache().getRegion("region1").close());

    if (useAccessor) {
      vm3.invoke(() -> getCache().getRegion("region1").close());
    }

    // We need to restart both VMs at the same time, because
    // they will wait for each other before allowing operations.
    AsyncInvocation<Void> createRegionOnVM0 = vm0.invokeAsync(
        () -> createPersistentPartitionedRegion("region1", getUniqueName(), getDiskDirs(),
            redundantCopies));
    AsyncInvocation<Void> createRegionOnVM2 = vm2.invokeAsync(
        () -> createPersistentPartitionedRegion("region1", getUniqueName(), getDiskDirs(),
            redundantCopies));

    createRegionOnVM0.await();
    createRegionOnVM2.await();

    if (useAccessor) {
      vm3.invoke(() -> createAccessor("region1", "ds1"));
    }

    // pause for async bucket recovery threads to finish their work. Otherwise
    // the rebalance op may think that the other member doesn't have buckets, then
    // ask it to create them and get a negative reply because it actually does
    // have the buckets, causing the test to fail

    // Try to rebalance again. This shouldn't do anything, because
    // we already recovered redundancy earlier.
    // Note that we don't check the primary balance. This rebalance
    // has a race with the redundancy recovery thread, which is also trying
    // to rebalance primaries. So this thread might end up rebalancing primaries,
    // or it might not.
    if (!simulate) {
      rebalanceVM.invoke(() -> {
        ResourceManager manager = getCache().getResourceManager();

        RebalanceResults results = doRebalance(simulate, manager);

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

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

        PartitionRebalanceInfo details = detailSet.iterator().next();
        assertThat(details.getBucketCreatesCompleted()).isEqualTo(0);
        assertThat(details.getBucketTransfersCompleted()).isEqualTo(0);

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

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

      // Redundancy should be repaired.
      vm0.invoke(() -> validateRedundancy("region1", 6, 1, 0));
    }

    vm1.invoke(
        () -> createPersistentPartitionedRegion("region1", getUniqueName(), getDiskDirs(),
            redundantCopies));

    // Look at vm0 buckets.
    assertThat(vm0.invoke(() -> getBucketList("region1"))).isEqualTo(bucketsOnVM0);

    /*
     * Look at vm1 buckets.
     */
    if (simulate) {
      /*
       * No rebalancing above because the simulation flag is on. Therefore, vm1 will have recovered
       * its buckets. We need to wait for the buckets because they might still be in the middle of
       * creation in the background
       */
      vm1.invoke(() -> waitForBucketList("region1", bucketsOnVM1));
    } else {
      /*
       * vm1 should have no buckets because offline buckets were recovered when vm0 and vm2 were
       * rebalanced above.
       */
      assertThat(vm1.invoke(() -> getBucketList("region1"))).isEmpty();
    }

    // look at vm2 buckets
    assertThat(vm2.invoke(() -> getBucketList("region1"))).isEqualTo(bucketsOnVM2);
  }
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