@Test
  public void testPartialKeyInPR() throws Exception {
    // set <true, false> means <PR=true, notifyBySubscription=false> to test local-invalidates
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(PARTITION_PERSISTENT)
        .create());
    int serverPort2 = server2.invoke(() -> new ServerBuilder()
        .regionShortcut(PARTITION_PERSISTENT)
        .create());

    for (VM clientVM : asList(client1, client2)) {
      clientVM.invoke(() -> new ClientBuilder()
          .serverPorts(serverPort1, serverPort2)
          .subscriptionAckInterval()
          .subscriptionEnabled(true)
          .subscriptionRedundancy()
          .create());
    }

    // server1 add slow listener
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator().addCacheListener(new SlowCacheListener<>());
    });

    // server2 add slow listener
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator()
          .addCacheListener(new SlowCountingCacheListener<>(new Action<>(Operation.CREATE,
              creates -> executorServiceRule.submit(() -> closeCacheConditionally(creates, 10)))));
    });

    // client2 add listener
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      region.getAttributesMutator().addCacheListener(new SlowCacheListener<>());
      region.registerInterest("ALL_KEYS");
    });

    // client1 add listener and putAll
    AsyncInvocation<Void> registerInterestAndPutAllInClient1 = client1.invokeAsync(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      region.getAttributesMutator().addCacheListener(new SlowCacheListener<>());
      region.registerInterest("ALL_KEYS");

      String message =
          String.format("Region %s putAll at server applied partial keys due to exception.",
              region.getFullPath());

      Throwable thrown = catchThrowable(() -> doPutAll(region, "Key-", ONE_HUNDRED));

      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasMessageContaining(message)
          .getCause()
          .isInstanceOf(PartitionOfflineException.class);
    });

    // server2 will closeCache after created 10 keys

    registerInterestAndPutAllInClient1.await();

    int sizeOnClient1 = client1.invoke(() -> getClientCache().getRegion(regionName).size());
    // client2Size maybe more than client1Size
    int sizeOnClient2 = client2.invoke(() -> getClientCache().getRegion(regionName).size());

    // restart server2
    server2.invoke(() -> {
      new ServerBuilder().regionShortcut(PARTITION_PERSISTENT).create();
    });

    await().untilAsserted(() -> {
      assertThat(server1.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(sizeOnClient2);
      assertThat(server2.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(sizeOnClient2);
    });

    // close a server to re-run the test
    server2.invoke(() -> getCache().close());
    int sizeOnServer1 = server1.invoke(() -> getCache().getRegion(regionName).size());

    // client1 does putAll again
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);

      String message =
          String.format("Region %s putAll at server applied partial keys due to exception.",
              region.getFullPath());

      Throwable thrown = catchThrowable(() -> doPutAll(region, "key-" + "again:", ONE_HUNDRED));

      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasMessageContaining(message)
          .getCause()
          .isInstanceOf(PartitionOfflineException.class);
    });

    int newSizeOnServer1 = server1.invoke(() -> getCache().getRegion(regionName).size());
    int newSizeOnClient1 = client1.invoke(() -> getClientCache().getRegion(regionName).size());
    int newSizeOnClient2 = client2.invoke(() -> getClientCache().getRegion(regionName).size());

    assertThat(newSizeOnServer1).isEqualTo(sizeOnServer1 + ONE_HUNDRED / 2);
    assertThat(newSizeOnClient1).isEqualTo(sizeOnClient1 + ONE_HUNDRED / 2);
    assertThat(newSizeOnClient2).isEqualTo(sizeOnClient2 + ONE_HUNDRED / 2);

    // restart server2
    server2.invoke(() -> {
      new ServerBuilder().regionShortcut(PARTITION_PERSISTENT).create();
    });
    sizeOnServer1 = server1.invoke(() -> getCache().getRegion(regionName).size());
    int sizeOnServer2 = server2.invoke(() -> getCache().getRegion(regionName).size());
    assertThat(sizeOnServer2).isEqualTo(sizeOnServer1);

    // server1 execute P2P putAll
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      // let the server to trigger exception after created 15 keys
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.CREATE,
              creates -> {
                if (creates >= 15) {
                  throw new CacheWriterException("Expected by test");
                }
              })));
    });

    // server2 add listener and putAll
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      Throwable thrown =
          catchThrowable(() -> doPutAll(region, "key-" + "once again:", ONE_HUNDRED));
      assertThat(thrown)
          .isInstanceOf(CacheWriterException.class)
          .hasMessageContaining("Expected by test");
    });

    newSizeOnServer1 = server1.invoke(() -> getCache().getRegion(regionName).size());
    int newSizeOnServer2 = server2.invoke(() -> getCache().getRegion(regionName).size());
    assertThat(newSizeOnServer1).isEqualTo(sizeOnServer1 + 15);
    assertThat(newSizeOnServer2).isEqualTo(sizeOnServer2 + 15);
  }
Tests partial key putAll to 2 PR servers, because putting data at server side is different
between PR and LR. PR does it in postPutAll. It's not running in singleHop putAll
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/PutAllClientServerDistributedTest.java