@Test
  public void recoversBucketsAfterRestartAfterCrashDuringGii() throws Exception {
    addIgnoredException(PartitionOfflineException.class);

    vm0.invoke(() -> createPartitionedRegion(1, -1, 113, true));

    vm0.invoke(() -> createData(0, 1, "value", partitionedRegionName));

    // Add an observer which will close the cache when the GII starts
    vm0.invoke(() -> {
      DistributionMessageObserver.setInstance(new DistributionMessageObserver() {
        @Override
        public void beforeProcessMessage(ClusterDistributionManager dm,
            DistributionMessage message) {
          if (message instanceof RequestImageMessage) {
            RequestImageMessage requestImageMessage = (RequestImageMessage) message;
            if (requestImageMessage.regionPath.contains("_0")) {
              DistributionMessageObserver.setInstance(null);
              getCache().close();
            }
          }
        }

      });
    });

    vm1.invoke(() -> createPartitionedRegion(1, -1, 113, true));

    // Make sure vm1 didn't create the bucket
    assertThat(vm1.invoke(() -> getBucketList(partitionedRegionName))).isEmpty();

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

    AsyncInvocation<Void> createPartitionedRegionOnVM0 =
        vm0.invokeAsync(() -> createPartitionedRegion(1, -1, 113, true));
    createPartitionedRegionOnVM0.join(SECONDS.toMillis(1));
    // vm0 should get stuck waiting for vm1 to recover from disk,
    // because vm0 thinks vm1 has the bucket
    assertThat(createPartitionedRegionOnVM0.isAlive()).isTrue();
    assertThat(createPartitionedRegionOnVM0).isNotCancelled().isNotDone();

    vm1.invoke(() -> createPartitionedRegion(1, -1, 113, true));

    // Make sure vm0 recovers the bucket
    assertThat(vm0.invoke(() -> getBucketList(partitionedRegionName))).containsOnly(0);

    // vm1 should satisfy redundancy for the bucket as well
    await().untilAsserted(() -> {
      assertThat(vm1.invoke(() -> getBucketList(partitionedRegionName))).containsOnly(0);
    });
  }
Another RegressionTest for bug 41436. If the GII source crashes before the GII is complete, we
need to make sure that later we can recover redundancy.

<p>
In this test case, we bring the GII down before we bring the source back up, to make sure the
source still discovers that the GII target is no longer hosting the bucket.

<p>
TRAC #41436: PR redundancy lost with HA (expected n members in primary list, but found n-1)
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionRegressionTest.java