@Test
  @Parameters({"true", "false"})
  @TestCaseName("{method}(concurrencyChecksEnabled={0})")
  public void test2NormalServer(boolean concurrencyChecksEnabled) {
    Properties config = getDistributedSystemProperties();
    config.setProperty(LOCATORS, locators);

    // set notifyBySubscription=false to test local-invalidates
    int serverPort1 = server1.invoke(() -> {
      createCache(new CacheFactory(config));
      return createServerRegion(regionName, concurrencyChecksEnabled);
    });
    int serverPort2 = server2.invoke(() -> {
      createCache(new CacheFactory(config));
      return createServerRegion(regionName, concurrencyChecksEnabled);
    });

    client1.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort1)
        .create());
    client2.invoke(() -> new ClientBuilder()
        .serverPorts(serverPort2)
        .create());

    // test case 1: putAll and removeAll to server1

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

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

      region.put(String.valueOf(ONE_HUNDRED), new TickerData());
    });

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

      region.localDestroy(String.valueOf(ONE_HUNDRED));
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);

      for (int i = 0; i < ONE_HUNDRED; i++) {
        TickerData tickerData = region.getEntry("case1-" + 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<>());
      // normal policy will not distribute create events
      assertThat(region.size()).isZero();
    });

    // client1 removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED + 1);

      // do removeAll with some keys not exist
      doRemoveAll(region, "case1-", ONE_HUNDRED * 2);
      assertThat(region.size()).isEqualTo(1);

      region.localDestroy(String.valueOf(ONE_HUNDRED));
    });

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

    // test case 2: putAll to server1, removeAll to server2

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

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

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

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

    // verify cache server 2
    server2.invoke(() -> {
      // normal policy will not distribute create events
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });

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

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

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

    // client1 verify removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(100));
      doRemoveAll(region, "case2-", ONE_HUNDRED);
    });

    // test case 3: removeAll a list with duplicated keys

    // put 3 keys then removeAll
    client1.invoke(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      region.put("case3-1", new TickerData());
      region.put("case3-2", new TickerData());
      region.put("case3-3", new TickerData());
      assertThat(region.size()).isEqualTo(3);

      Collection<String> keys = new ArrayList<>();
      keys.add("case3-1");
      keys.add("case3-2");
      keys.add("case3-3");
      keys.add("case3-1");
      region.removeAll(keys);
      assertThat(region.size()).isZero();
    });

    // verify cache server 1
    server1.invoke(() -> {
      assertThat(getCache().getRegion(regionName).size()).isZero();
    });
  }
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