@Test
  @Parameters({"true", "false"})
  @TestCaseName("{method}(singleHop={0})")
  public void clientPutAllIsMissingKeyUntilShutdownServerRestarts(boolean singleHop) {
    int serverPort1 = server1.invoke(() -> new ServerBuilder()
        .regionShortcut(PARTITION_PERSISTENT)
        .create());
    server2.invoke(() -> new ServerBuilder()
        .regionShortcut(PARTITION_PERSISTENT)
        .create());

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

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

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

      // do putAll to create all buckets
      doPutAll(region, "key-", ONE_HUNDRED);
      assertThat(region.size()).isEqualTo(ONE_HUNDRED);
    });

    server2.invoke(() -> getCache().close());

    // putAll from client 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, keyPrefix, ONE_HUNDRED));

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

      assertThat(region.size()).isEqualTo(ONE_HUNDRED * 3 / 2);

      int count = 0;
      for (int i = 0; i < ONE_HUNDRED; i++) {
        VersionTag<?> tag = ((InternalRegion) region).getVersionTag(keyPrefix + i);
        if (tag != null && 1 == tag.getEntryVersion()) {
          count++;
        }
      }

      assertThat(count).isEqualTo(ONE_HUNDRED / 2);

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

      thrown = catchThrowable(() -> doRemoveAll(region, keyPrefix, ONE_HUNDRED));

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

      // putAll only created 50 entries, removeAll removed them. So 100 entries are all NULL
      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry(keyPrefix + i);
        assertThat(regionEntry).isNull();
      }
    });

    // verify entries from client2
    client2.invoke(() -> await().untilAsserted(() -> {
      Region<String, TickerData> region = getClientCache().getRegion(regionName);
      for (int i = 0; i < ONE_HUNDRED; i++) {
        Entry<String, TickerData> regionEntry = region.getEntry(keyPrefix + i);
        assertThat(regionEntry).isNull();
      }
    }));
  }
Bug: Data inconsistency between client/server with HA (Object is missing in client cache)

<p>
This is a known issue for persist region with HA.
<ul>
<li>client1 sends putAll key1,key2 to server1
<li>server1 applied the key1 locally, but the putAll failed due to CacheClosedException? (HA
test).
<li>Then client1 will not have key1. But when server1 is restarted, it recovered key1 from
persistence.
<li>Then data mismatch btw client and server.
</ul>

<p>
There's a known issue which can be improved by changing the test:
<ul>
<li>edge_14640 sends a putAll to servers. But the server is shutdown due to HA test. However at
that time, the putAll might have put the key into local cache.
<li>When the server is restarted, the servers will have the key, other clients will also have
the key (due to register interest), but the calling client (i.e. edge_14640) will not have the
key, unless it's restarted.
</ul>

<pre>
In above scenario, even changing redundantCopies to 1 will not help. This is the known issue. To improve the test, we should let all the clients restart before doing verification.
How to verify the result falls into above scenario?
1) errors.txt, the edge_14640 has inconsistent cache compared with server's snapshot.
2) grep Object_82470 *.log and found the mismatched key is originated from 14640.
</pre>
geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/PutAllClientServerDistributedTest.java