@Test
  public void testPartialKeyInPRSingleHop() 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());

    client1.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort1, serverPort2)
        .create());
    client2.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort1, serverPort2)
        .subscriptionAckInterval()
        .subscriptionEnabled(true)
        .subscriptionRedundancy()
        .create());

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

    // do some putAll to get ClientMetaData for future putAll
    client1.invoke(() -> doPutAll(getClientCache().getRegion(regionName), "key-", ONE_HUNDRED));

    await().untilAsserted(() -> {
      assertThat(client1.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(ONE_HUNDRED);
      assertThat(client2.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(ONE_HUNDRED);
      assertThat(server1.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(ONE_HUNDRED);
      assertThat(server2.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(ONE_HUNDRED);
    });

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

    // add a listener that will close the cache at the 10th update

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

    // client1 add listener and putAll
    AsyncInvocation<Void> addListenerAndPutAllInClient1 = client1.invokeAsync(() -> {
      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, keyPrefix, ONE_HUNDRED));

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

    // server2 will closeCache after creating 10 keys

    addListenerAndPutAllInClient1.await();

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

    // Test Case1: Trigger singleHop putAll. Stop server2 in middle.
    // ONE_HUNDRED_ENTRIES/2 + X keys will be created at servers. i.e. X keys at server2,
    // ONE_HUNDRED_ENTRIES/2 keys at server1.
    // The client should receive a PartialResultException due to PartitionOffline

    // close a server to re-run the test
    server2.invoke(() -> getCache().close());

    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, keyPrefix + "again:", ONE_HUNDRED));

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

    // Test Case 2: based on case 1, but this time, there should be no X keys
    // created on server2.

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

    // add a cacheWriter for server to fail putAll after it created cacheWriterAllowedKeyNum keys
    int throwAfterNumberCreates = 16;

    // server1 add cacheWriter to throw exception after created some keys
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.CREATE,
              creates -> {
                if (creates >= throwAfterNumberCreates) {
                  throw new CacheWriterException("Expected by test");
                }
              })));
    });

    // client1 does putAll once more
    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, keyPrefix + "once more:", ONE_HUNDRED));

      assertThat(thrown)
          .isInstanceOf(ServerOperationException.class)
          .hasMessageContaining(message)
          .getCause()
          .isInstanceOf(CacheWriterException.class)
          .hasMessageContaining("Expected by test");
    });
  }
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. This is a singlehop putAll test.
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/PutAllClientServerDistributedTest.java