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

    // Create the region in only 1 VM
    vm0.invoke(() -> createPartitionedRegion("region1", 0));

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

    // Create the region in the other VM (should have no effect)
    vm1.invoke(() -> createPartitionedRegion("region1", 0));

    // Now do a rebalance, but start another member in the middle
    vm0.invoke(() -> {
      InternalResourceManager manager = getCache().getInternalResourceManager();

      InternalResourceManager.setResourceObserver(new ResourceObserverAdapter() {
        private volatile boolean firstBucket = true;

        @Override
        public void movingBucket(Region region, int bucketId, DistributedMember source,
            DistributedMember target) {
          if (firstBucket) {
            firstBucket = false;
            // NOTE: do not replace createPartitionedRegionRunnable with lambda
            // because ResourceObserverAdapter is NOT serializable
            vm2.invoke(createPartitionedRegionRunnable("region1", 0));
          }
        }
      });

      RebalanceResults results = doRebalance(false, manager);

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

      Set<PartitionMemberInfo> beforeDetails = details.getPartitionMemberDetailsBefore();
      // there should have only been 2 members when the rebalancing started.
      assertThat(beforeDetails).hasSize(2);

      // if it was done, there should now be 3 members.
      Set<PartitionMemberInfo> afterDetails = details.getPartitionMemberDetailsAfter();
      assertThat(afterDetails).hasSize(3);

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

      validateStatistics(manager, results);

      ResourceManagerStats stats = manager.getStats();
      assertThat(stats.getRebalanceMembershipChanges()).isEqualTo(1);
    });
  }
Test that the rebalancing operation picks up on a concurrent membership change
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/control/RebalanceOperationDistributedTest.java