@Test
  @Parameters({"PUT", "INVALIDATE", "DESTROY", "CACHE_LOADER"})
  @TestCaseName("{method}(operation={0})")
  public void runTestWaitForOperation(OperationEnum operation) throws Exception {
    VM vm1 = getVM(1);

    // Create a region in this VM with a cache writer
    // and cache loader
    createPartitionedRegion("region1", 1, 100, (CacheLoader<Number, String>) helper1 -> "anobject");
    Region<Number, String> region = getCache().getRegion("region1");

    // create some buckets
    region.put(1, "A");
    region.put(2, "A");

    BlockingCacheListener<Number, String> cacheWriter = new BlockingCacheListener<>(2);
    region.getAttributesMutator().addCacheListener(cacheWriter);

    // start two threads doing operations, one on each bucket
    // the threads will block on the cache writer. The rebalance operation
    // will try to move one of these buckets, but it shouldn't
    // be able to because of the in progress operation.
    executorServiceRule.submit(() -> operation.execute(region, 1));
    executorServiceRule.submit(() -> operation.execute(region, 2));

    cacheWriter.waitForOperationsStarted();

    // make sure we can tell that the buckets have low redundancy
    validateRedundancy("region1", 2, 0, 2);

    // Create the region in the other VM (should have no effect)
    vm1.invoke(() -> createPartitionedRegion("region1", 1, 100,
        (CacheLoader<Number, String>) helper1 -> "anobject"));

    // Make sure we still have low redundancy
    validateRedundancy("region1", 2, 0, 2);

    ResourceManager manager = getCache().getResourceManager();
    RebalanceOperation rebalance = manager.createRebalanceFactory().start();

    Throwable thrown = catchThrowable(() -> rebalance.getResults(5, SECONDS));
    assertThat(thrown).isInstanceOf(TimeoutException.class);

    cacheWriter.release();

    RebalanceResults results = rebalance.getResults(TIMEOUT_SECONDS, SECONDS);
    assertThat(results.getTotalBucketCreatesCompleted()).isEqualTo(2);
    assertThat(results.getTotalPrimaryTransfersCompleted()).isEqualTo(1);
    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(2);
    assertThat(details.getPrimaryTransfersCompleted()).isEqualTo(1);
    assertThat(details.getBucketTransferBytes()).isEqualTo(0);
    assertThat(details.getBucketTransfersCompleted()).isEqualTo(0);

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

    for (PartitionMemberInfo memberDetails : afterDetails) {
      assertThat(memberDetails.getBucketCount()).isEqualTo(2);
      assertThat(memberDetails.getPrimaryCount()).isEqualTo(1);

      validateRedundancy("region1", 2, 1, 0);
      vm1.invoke(() -> validateRedundancy("region1", 2, 1, 0));
    }
  }
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