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

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

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

    Set<Integer> bucketsOnVM0 = vm0.invoke(this::getBucketList);
    Set<Integer> bucketsLost = vm1.invoke(this::getBucketList);

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

    // VM2 should pick up the slack

    await().until(() -> {
      Set<Integer> vm2Buckets = vm2.invoke(this::getBucketList);
      return bucketsLost.equals(vm2Buckets);
    });

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

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

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

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

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

    // Make sure we wait for vm2, because it's got the latest copy of the bucket
    createPartitionedRegionOnVM1.join(SECONDS.toMillis(1));
    assertThat(createPartitionedRegionOnVM1.isAlive()).isTrue();

    AsyncInvocation<Void> createPartitionedRegionOnVM2 =
        vm2.invokeAsync(() -> createPartitionedRegion(1, 0, 113, true));

    createPartitionedRegionOnVM2.await();
    createPartitionedRegionOnVM0.await();
    createPartitionedRegionOnVM1.await();

    // The data shouldn't be affected.
    vm1.invoke(() -> checkData(0, NUM_BUCKETS, "b"));
    assertThat(vm1.invoke(this::getBucketList)).isEmpty();
    assertThat(vm0.invoke(this::getBucketList)).isEqualTo(bucketsOnVM0);
    assertThat(vm2.invoke(this::getBucketList)).isEqualTo(bucketsOnVM0);
  }
Test to make sure that we recreate a bucket if a member is revoked, and that we do it
immediately if recovery delay is set to 0.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionDistributedTest.java