@Test
  public void recoversAfterBucketCreationCrashes() throws Exception {
    vm0.invoke(() -> {
      DistributionMessageObserver.setInstance(new DistributionMessageObserver() {
        @Override
        public void beforeSendMessage(ClusterDistributionManager dm, DistributionMessage message) {
          if (message instanceof ManageBucketMessage.ManageBucketReplyMessage) {
            Cache cache = getCache();
            disconnectFromDS();
            await().until(cache::isClosed);
            CRASHED.set(true);
          }
        }
      });
    });

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

    vm1.invoke(() -> createData(0, 4, "a", partitionedRegionName));

    Set<Integer> bucketsOnVM1 = vm1.invoke(() -> getBucketList(partitionedRegionName));
    assertThat(bucketsOnVM1).hasSize(4);

    vm1.invoke(() -> checkData(0, 4, "a", partitionedRegionName));

    // wait till cache is completely shutdown before trying to create the region again. otherwise
    // deadlock situation might happen.
    vm0.invoke(() -> await().until(CRASHED::get));
    vm0.invoke(() -> createPartitionedRegion(0, -1, 113, true));
    vm0.invoke(() -> checkData(0, 4, "a", partitionedRegionName));

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

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

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

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

    vm0.invoke(() -> checkData(0, 4, "a", partitionedRegionName));
    assertThat(vm1.invoke(() -> getBucketList(partitionedRegionName))).isEqualTo(bucketsOnVM1);
    assertThat(vm0.invoke(() -> getBucketList(partitionedRegionName))).isEmpty();
  }
RegressionTest for bug 41336

<p>
TRAC #41336: Unexpected PartitionOfflineException
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionRegressionTest.java