@Test
  public void testCrashDuringRedundancySatisfaction() throws Exception {
    vm0.invoke(() -> {
      createCache();
      createDiskStore(diskStoreName1);
      createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY, 1, -1);
      createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
          DEFAULT_RECOVERY_DELAY, 1, -1);

      createData(regionName, "a");
      createData(childRegionName1, "a");
    });

    vm1.invoke(() -> {
      createCache();
      createDiskStore(diskStoreName1);
      createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY, 1, -1);
      createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
          DEFAULT_RECOVERY_DELAY, 1, -1);
    });

    // We shouldn't have created any buckets in vm1 yet.
    vm1.invoke(() -> {
      assertThat(getBucketIds(regionName)).isEmpty();
    });

    // Add an observer that will disconnect before allowing the peer to GII a colocated bucket.
    // This should leave the peer with only the parent bucket
    vm0.invoke(() -> {
      latch = new CountDownLatch(1);

      DistributionMessageObserver.setInstance(new DistributionMessageObserver() {
        @Override
        public void beforeProcessMessage(ClusterDistributionManager dm,
            DistributionMessage message) {
          if (message instanceof RequestImageMessage) {
            RequestImageMessage requestImageMessage = (RequestImageMessage) message;
            if (requestImageMessage.getRegionPath().contains(regionName) ||
                requestImageMessage.getRegionPath().contains(childRegionName1)) {
              DistributionMessageObserver.setInstance(null);

              latch.countDown();
            }
          }
        }
      });
    });

    AsyncInvocation<Void> disconnectDuringGiiInVm0 = vm0.invokeAsync(() -> {
      latch.await(TIMEOUT_MILLIS, MILLISECONDS);

      closeCache();
    });

    vm1.invoke(() -> {
      try (IgnoredException ie = addIgnoredException(PartitionOfflineException.class)) {
        // Do a rebalance to create buckets in vm1. THis will cause vm0 to disconnect
        // as we satisfy redundancy with vm1.
        Throwable thrown = catchThrowable(() -> {
          getCache().getResourceManager().createRebalanceFactory().start().getResults();
        });
        if (thrown != null) {
          assertThat(thrown).isInstanceOf(PartitionOfflineException.class);
        }
      }
    });

    disconnectDuringGiiInVm0.await();

    // close the cache in vm1
    vm1.invoke(this::closeCache);

    // Create the cache and PRs on both members
    for (VM vm : toArray(vm0, vm1)) {
      addAsync(vm.invokeAsync(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY, 1, -1);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, 1, -1);
      }));
    }
    awaitAllAsync();

    // Make sure the data was recovered correctly
    vm0.invoke(() -> {
      validateData(regionName, "a");
      validateData(childRegionName1, "a");
    });
  }
Test what happens when we crash in the middle of satisfying redundancy for a colocated bucket.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentColocatedPartitionedRegionDistributedTest.java