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

    DistributedMember member1 = vm0.invoke(() -> {
      createPartitionedRegion("region1", 1, 100);
      return getCache().getDistributedSystem().getDistributedMember();
    });

    vm0.invoke(() -> {
      Region<Number, String> region = getCache().getRegion("region1");
      for (int i = 0; i < 1; i++) {
        region.put(i, "A");
      }
    });

    vm0.invoke(() -> validateRedundancy("region1", 1, 0, 1));

    // Now create the region in 2 more VMs
    // Let localMaxMemory(VM1) > localMaxMemory(VM2)
    // so that redundant bucket will always be attempted on VM1
    DistributedMember member2 = vm1.invoke(() -> {
      createPartitionedRegion("region1", 1, 100);
      return getCache().getDistributedSystem().getDistributedMember();
    });
    DistributedMember member3 = vm2.invoke(() -> {
      createPartitionedRegion("region1", 1, 90);
      return getCache().getDistributedSystem().getDistributedMember();
    });

    vm0.invoke(() -> validateRedundancy("region1", 1, 0, 1));

    // Inject mock PRHARedundancyProvider to simulate createBucketFailures
    vm0.invoke(() -> {
      InternalCache cache = spy(getCache());

      PartitionedRegion origRegion = (PartitionedRegion) cache.getRegion("region1");
      PartitionedRegion spyRegion = spy(origRegion);
      PRHARedundancyProvider redundancyProvider =
          spy(new PRHARedundancyProvider(spyRegion, mock(InternalResourceManager.class)));

      // return the spied region when ever getPartitionedRegions() is invoked
      Set<PartitionedRegion> parRegions = cache.getPartitionedRegions();
      parRegions.remove(origRegion);
      parRegions.add(spyRegion);

      doReturn(parRegions).when(cache).getPartitionedRegions();
      doReturn(redundancyProvider).when(spyRegion).getRedundancyProvider();

      // simulate create bucket fails on member2 and test if it creates on member3
      doReturn(false).when(redundancyProvider).createBackupBucketOnMember(anyInt(),
          eq((InternalDistributedMember) member2), anyBoolean(), anyBoolean(), any(),
          anyBoolean());

      // Now simulate a rebalance
      // Create operationImpl and not factory as we need spied cache to be passed to operationImpl
      RegionFilter filter = new FilterByPath(null, null);
      RebalanceOperationImpl operation = new RebalanceOperationImpl(cache, false, filter);
      operation.start();

      RebalanceResults results = operation.getResults(TIMEOUT_SECONDS, SECONDS);

      assertThat(results.getTotalBucketCreatesCompleted()).isEqualTo(1);
      assertThat(results.getTotalPrimaryTransfersCompleted()).isEqualTo(0);
      assertThat(results.getTotalBucketTransferBytes()).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(1);
      assertThat(details.getPrimaryTransfersCompleted()).isEqualTo(0);
      assertThat(details.getBucketTransferBytes()).isEqualTo(0);
      assertThat(details.getBucketTransfersCompleted()).isEqualTo(0);

      Set<PartitionMemberInfo> afterDetails = details.getPartitionMemberDetailsAfter();
      assertThat(afterDetails).hasSize(3);
      for (PartitionMemberInfo memberDetails : afterDetails) {
        if (memberDetails.getDistributedMember().equals(member1)) {
          assertThat(memberDetails.getBucketCount()).isEqualTo(1);
          assertThat(memberDetails.getPrimaryCount()).isEqualTo(1);
        } else if (memberDetails.getDistributedMember().equals(member2)) {
          assertThat(memberDetails.getBucketCount()).isEqualTo(0);
          assertThat(memberDetails.getPrimaryCount()).isEqualTo(0);
        } else if (memberDetails.getDistributedMember().equals(member3)) {
          assertThat(memberDetails.getBucketCount()).isEqualTo(1);
          assertThat(memberDetails.getPrimaryCount()).isEqualTo(0);
        }
      }

      ResourceManagerStats stats = cache.getInternalResourceManager().getStats();

      assertThat(stats.getRebalancesInProgress()).isEqualTo(0);
      assertThat(stats.getRebalancesCompleted()).isEqualTo(1);
      assertThat(stats.getRebalanceBucketCreatesInProgress()).isEqualTo(0);
      assertThat(stats.getRebalanceBucketCreatesCompleted())
          .isEqualTo(results.getTotalBucketCreatesCompleted());
      assertThat(stats.getRebalanceBucketCreatesFailed()).isEqualTo(1);
    });

    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> validateRedundancy("region1", 1, 1, 0));
    }
  }
Test that we correctly use the redundancy-zone property to determine where to place redundant
copies of a buckets.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java