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

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

    // server1 add cacheWriter
    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");
                }
              })));
    });

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

    // client1 putAll
    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-", ONE_HUNDRED));

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

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

    int sizeOnServer1 = server1.invoke(() -> getCache().getRegion(regionName).size());

    // server1 add cacheWriter
    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-" + "again:", ONE_HUNDRED));
      assertThat(thrown).isInstanceOf(CacheWriterException.class);
    });

    int sizeOnServer2 = server2.invoke(() -> getCache().getRegion(regionName).size());
    assertThat(sizeOnServer2).isEqualTo(sizeOnServer1 + 15);

    await().untilAsserted(() -> {
      // client 1 did not register interest
      assertThat(client1.invoke(() -> getClientCache().getRegion(regionName).size())).isEqualTo(15);
      assertThat(client2.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2);
      assertThat(server1.invoke(() -> getCache().getRegion(regionName).size())).isEqualTo(15 * 2);
      assertThat(server2.invoke(() -> getCache().getRegion(regionName).size())).isEqualTo(15 * 2);
    });

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

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

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

      Throwable thrown = catchThrowable(() -> doRemoveAll(region, "key-", ONE_HUNDRED));

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

    await().untilAsserted(() -> {
      // client 1 did not register interest
      assertThat(client1.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(15 - 5);
      assertThat(client2.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2 - 5);
      assertThat(server1.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2 - 5);
      assertThat(server2.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2 - 5);
    });

    // server1 add cacheWriter
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      // server triggers exception after destroying 5 keys
      region.getAttributesMutator()
          .setCacheWriter(new ActionCacheWriter<>(new Action<>(Operation.DESTROY,
              ops -> {
                if (ops >= 5) {
                  throw new CacheWriterException("Expected by test");
                }
              })));
    });

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

    await().untilAsserted(() -> {
      // client 1 did not register interest
      assertThat(client1.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(15 - 5);
      assertThat(client2.invoke(() -> getClientCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2 - 5 - 5);
      assertThat(server1.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2 - 5 - 5);
      assertThat(server2.invoke(() -> getCache().getRegion(regionName).size()))
          .isEqualTo(15 * 2 - 5 - 5);
    });
  }
Tests partial key putAll and removeAll to 2 servers with local region
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/PutAllClientServerDistributedTest.java