@Test
  public void bucketsRecoverAfterRestartingOriginalServer() throws Exception {
    vm0.invoke(() -> createPartitionedRegion(1, 0, 113, true));
    vm1.invoke(() -> createPartitionedRegion(1, 0, 113, true));

    vm0.invoke(() -> createData(0, NUM_BUCKETS, "a", partitionedRegionName));

    // This should do nothing because we have satisfied redundancy.
    vm2.invoke(() -> createPartitionedRegion(1, 0, 113, true));
    assertThat(vm2.invoke(() -> getBucketList(partitionedRegionName))).isEmpty();

    Set<Integer> bucketsOnVM0 = vm0.invoke(() -> getBucketList(partitionedRegionName));
    Set<Integer> bucketsLost = vm1.invoke(() -> getBucketList(partitionedRegionName));

    vm1.invoke(() -> getCache().close());

    // VM2 should pick up the slack
    vm2.invoke(() -> awaitBucketRecovery(bucketsLost));

    vm0.invoke(() -> createData(0, NUM_BUCKETS, "b", partitionedRegionName));

    // VM1 should recover, but it shouldn't host the bucket anymore
    vm1.invoke(() -> createPartitionedRegion(1, 0, 113, true));

    // The data shouldn't be affected.
    vm1.invoke(() -> checkData(0, NUM_BUCKETS, "b", partitionedRegionName));

    vm2.invoke(() -> getCache().close());

    // The buckets should move back to vm1.
    vm1.invoke(() -> awaitBucketRecovery(bucketsLost));

    assertThat(vm0.invoke(() -> getBucketList(partitionedRegionName))).isEqualTo(bucketsOnVM0);
    assertThat(vm1.invoke(() -> getBucketList(partitionedRegionName))).isEqualTo(bucketsOnVM0);

    // The data shouldn't be affected.
    vm1.invoke(() -> checkData(0, NUM_BUCKETS, "b", partitionedRegionName));

    // restart everything, and make sure it comes back correctly.

    vm0.invoke(() -> getCache().close());
    vm1.invoke(() -> getCache().close());

    AsyncInvocation<Void> createPartitionedRegionOnVM0 =
        vm0.invokeAsync(() -> createPartitionedRegion(1, -1, 113, true));
    AsyncInvocation<Void> createPartitionedRegionOnVM1 =
        vm1.invokeAsync(() -> createPartitionedRegion(1, -1, 113, true));

    createPartitionedRegionOnVM0.await(2, MINUTES);
    createPartitionedRegionOnVM1.await(2, MINUTES);

    // The data shouldn't be affected.
    vm1.invoke(() -> checkData(0, NUM_BUCKETS, "b", partitionedRegionName));
    assertThat(vm0.invoke(() -> getBucketList(partitionedRegionName))).isEqualTo(bucketsOnVM0);
    assertThat(vm1.invoke(() -> getBucketList(partitionedRegionName))).isEqualTo(bucketsOnVM0);
  }
RegressionTest: replace buckets which are offline on A by creating them on C. We then
shutdown C and restart A, which recovers those buckets.

<p>
TRAC 41340: data inconsistency after disk recovery from persistent PR
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionRegressionTest.java