@Test
  public void test2Server() throws Exception {
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());
    int serverPort2 = server2.invoke(() -> new ServerBuilder()
        .regionShortcut(REPLICATE)
        .create());

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

    // client1 putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doPutAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // verify cache server 1
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator().setCacheWriter(new CountingCacheWriter<>());
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // verify cache server 2
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      region.getAttributesMutator().setCacheWriter(new CountingCacheWriter<>());
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
      }
    });

    // client2 verify putAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("key-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // client1 removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doRemoveAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
    });

    // verify removeAll cache server 1
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isZero();

      CountingCacheWriter<String, TickerData> countingCacheWriter =
          (CountingCacheWriter<String, TickerData>) region.getAttributes().getCacheWriter();
      assertThat(countingCacheWriter.getDestroys()).isEqualTo(ONE_HUNDRED);
    });

    // verify removeAll cache server 2
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isZero();

      CountingCacheWriter<String, TickerData> countingCacheWriter =
          (CountingCacheWriter<String, TickerData>) region.getAttributes().getCacheWriter();
      // beforeDestroys are only triggered at server1 since removeAll is submitted from client1
      assertThat(countingCacheWriter.getDestroys()).isZero();
    });

    // client2 verify removeAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // async putAll1 from client1
    AsyncInvocation<Void> putAll1InClient1 = client1.invokeAsync(
        () -> doPutAll(getClientCache().getRegion(regionName), "async1key-", ONE_HUNDRED));

    // async putAll2 from client1
    AsyncInvocation<Void> putAll2InClient1 = client1.invokeAsync(
        () -> doPutAll(getClientCache().getRegion(regionName), "async2key-", ONE_HUNDRED));

    putAll1InClient1.await();
    putAll2InClient1.await();

    // verify client 1 for async keys
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2);

      long timeStamp1 = 0;
      long timeStamp2 = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("async1key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp1);
        timeStamp1 = tickerData.getTimeStamp();

        tickerData = region.getEntry("async2key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp2);
        timeStamp2 = tickerData.getTimeStamp();
      }
    });

    // verify cache server 1 for async keys
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2);

      long timeStamp1 = 0;
      long timeStamp2 = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("async1key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp1);
        timeStamp1 = tickerData.getTimeStamp();

        tickerData = region.getEntry("async2key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp2);
        timeStamp2 = tickerData.getTimeStamp();
      }
    });

    // verify cache server 2 for async keys
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2);

      long timeStamp1 = 0;
      long timeStamp2 = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("async1key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp1);
        timeStamp1 = tickerData.getTimeStamp();

        tickerData = region.getEntry("async2key-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp2);
        timeStamp2 = tickerData.getTimeStamp();
      }
    });

    // client2 verify async putAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED * 2));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("async1key-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("async2key-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // async removeAll1 from client1
    AsyncInvocation<Void> removeAll1InClient1 = client1.invokeAsync(() -> {
      doRemoveAll(getClientCache().getRegion(regionName), "async1key-", ONE_HUNDRED);
    });

    // async removeAll2 from client1
    AsyncInvocation<Void> removeAll2InClient1 = client1.invokeAsync(() -> {
      doRemoveAll(getClientCache().getRegion(regionName), "async2key-", ONE_HUNDRED);
    });

    removeAll1InClient1.await();
    removeAll2InClient1.await();

    // client1 removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      doRemoveAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
    });

    // verify async removeAll cache server 1
    server1.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

    // verify async removeAll cache server 2
    server2.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

    // client2 verify async removeAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // server1 execute P2P putAll
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      doPutAll(region, "p2pkey-", ONE_HUNDRED);

      long timeStamp = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("p2pkey-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp);
        timeStamp = tickerData.getTimeStamp();
      }
    });

    // verify cache server 2 for p2p keys
    server2.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);

      long timeStamp = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("p2pkey-" + i).getValue();
        assertThat(tickerData.getPrice()).isEqualTo(i);
        assertThat(tickerData.getTimeStamp()).isGreaterThanOrEqualTo(timeStamp);
        timeStamp = tickerData.getTimeStamp();
      }
    });

    // client2 verify p2p putAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("p2pkey-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // client1 verify p2p putAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(ONE_HUNDRED));

      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry("p2pkey-" + i);
        assertThat(regionEntry.getValue()).isNull();
      }
    });

    // server1 execute P2P removeAll
    server1.invoke(() -> {
      Region<String, TickerData> region = getCache().getRegion(regionName);
      doRemoveAll(region, "p2pkey-", ONE_HUNDRED);
      assertThat(region.size()).isZero();
    });

    // verify p2p removeAll cache server 2
    server2.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

    // client2 verify p2p removeAll
    client2.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // client1 verify p2p removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isZero());
    });

    // execute putAll on client2 for key 0-10
    client2.invoke(() -> doPutAll(getClientCache().getRegion(regionName), "key-", 10));

    // verify client1 for local invalidate
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      for (int i = 0; i < 10; i++) {
        String key = "key-" + i;

        await("entry with null value exists for " + key).until(() -> {
          Entry<String, TickerData> regionEntry = region.getEntry(key);
          return regionEntry != null && regionEntry.getValue() == null;
        });

        // local invalidate will set the value to null
        TickerData tickerData = region.getEntry("key-" + i).getValue();
        assertThat(tickerData).isNull();
      }
    });
  }
Tests putAll and removeAll to 2 servers. Use Case: 1) putAll from a single-threaded client to a
replicated region 2) putAll from a multi-threaded client to a replicated region 3)
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/PutAllClientServerDistributedTest.java