@Test

  public void testCopyOnReadWithBridgeServer() {
    System.setProperty(SystemProperty.DEFAULT_PREFIX + EARLY_ENTRY_EVENT_SERIALIZATION, "true");
    Invoke.invokeInEveryVM(
        () -> System.setProperty(SystemProperty.DEFAULT_PREFIX + EARLY_ENTRY_EVENT_SERIALIZATION,
            "true"));

    createBridgeServer(server, rName, ports[0]);
    // Put an instance of SerializationCounter to assert copy-on-read behavior
    // when notifyBySubscription is true
    server.invoke("Enable copy on read and assert server copy behavior", () -> {
      final Region<Object, Object> rootRegion = getRootRegion(rName);

      // Using a key that counts serialization, the test captures
      // any serialization of the key when it is a member of another object,
      // specifically in this case ClientUpdateMessageImpl which is assume to be
      // the value of a HARegion
      SerializationCountingKey key = new SerializationCountingKey(k1);
      byte[] val = new byte[1];
      byte valIsObj = 0x01;
      Integer cb = 0;
      ClientProxyMembershipID cpmi = null;
      EventID eid = null;
      ClientUpdateMessageImpl cui = new ClientUpdateMessageImpl(
          EnumListenerEvent.AFTER_CREATE, (LocalRegion) rootRegion, key, val, valIsObj, cb, cpmi,
          eid);
      ClientUpdateMessageImpl cuiCopy = CopyHelper.copy(cui);
      assertThat(cui.getKeyOfInterest()).isSameAs(key);
      assertThat(key.count.get()).isEqualTo(1);
      key = (SerializationCountingKey) cuiCopy.getKeyOfInterest();
      assertThat(cuiCopy.getKeyOfInterest()).isEqualTo(cui.getKeyOfInterest());
      assertThat(key.count.get()).isEqualTo(1);

      SerializationCountingKey ks1 = new SerializationCountingKey(k1);
      // AbstractRegionMap basicPut now serializes newValue in EntryEventImpl
      // which can be used for delivering client update message later
      SerializationCountingValue sc = new SerializationCountingValue();
      rootRegion.put(ks1, sc);
      assertThat(sc.count.get()).isEqualTo(1);
      assertThat(ks1.count.get()).isEqualTo(0);


      // No copy should be made upon get (assert standard no copy behavior)
      sc = (SerializationCountingValue) rootRegion.get(ks1);
      assertThat(sc.count.get()).isEqualTo(1);
      assertThat(ks1.count.get()).isEqualTo(0);


      // enable copy on read
      getCache().setCopyOnRead(true);

      // Assert standard copy on read behavior and basicPut in AbstractRegionMap
      sc = (SerializationCountingValue) rootRegion.get(ks1);
      assertThat(sc.count.get()).isEqualTo(2);
      assertThat(ks1.count.get()).isEqualTo(0);
      // Put another counter with copy-on-read true
      // AbstractRegionMap basicPut now serializes newValue
      sc = new SerializationCountingValue();
      SerializationCountingKey ks3 = new SerializationCountingKey(k3);
      rootRegion.put(ks3, sc);
      assertThat(sc.count.get()).isEqualTo(1);
      assertThat(ks3.count.get()).isEqualTo(0);
    });

    // Setup a client which subscribes to the server region, registers (aka pulls)
    // interest in keys which creates an assumed HARegionQueue on the server
    // (in the event that the above code didn't already create a HARegion)
    final String serverHostName = NetworkUtils.getServerHostName();
    client.invoke("Assert server copy behavior from client", () -> {
      getCache();

      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      ClientServerTestCase.configureConnectionPool(factory, serverHostName, ports, true, -1, 1,
          null);
      factory.setScope(Scope.LOCAL);
      Region<Object, Object> rootRegion = createRootRegion(rName, factory);
      SerializationCountingKey ks1 = new SerializationCountingKey(k1);
      SerializationCountingKey ks3 = new SerializationCountingKey(k3);
      // original two serializations on server and one serialization for register interest
      rootRegion.registerInterest(ks1, InterestResultPolicy.KEYS_VALUES);
      // entry shouldn't exist yet
      rootRegion.registerInterest(new SerializationCountingKey(k2),
          InterestResultPolicy.KEYS_VALUES);
      // original one serializations on server and one serialization for register interest
      rootRegion.registerInterest(ks3, InterestResultPolicy.KEYS_VALUES);

      // get from local cache.
      // original two serializations on server and one for previous register interest
      SerializationCountingValue sc = (SerializationCountingValue) rootRegion.get(ks1);
      assertThat(sc.count.get()).isEqualTo(3);

      sc = (SerializationCountingValue) rootRegion.get(ks3);
      assertThat(sc.count.get()).isEqualTo(2);
    });

    // Put an instance of SerializationCounter to assert copy-on-read behavior
    // once a client has registered interest
    server.invoke("Assert copy behavior after client is setup", () -> {
      Region<Object, Object> rootRegion = getRootRegion(rName);
      CacheServerImpl bsi = (CacheServerImpl) getCache().getCacheServers().iterator().next();
      Collection cp = bsi.getAcceptor().getCacheClientNotifier().getClientProxies();
      // Should only be one because only one client is connected
      assertThat(cp.size()).isEqualTo(1);
      final CacheClientProxy ccp = (CacheClientProxy) cp.iterator().next();
      // Wait for messages to drain to capture a stable "processed message count"
      await("region queue never became empty")
          .until(() -> ccp.getHARegionQueue().size() == 0);

      // Capture the current processed message count to know
      // when the next message has been serialized
      final int currMesgCount = ccp.getStatistics().getMessagesProcessed();

      SerializationCountingKey ks2 = new SerializationCountingKey(k2);
      SerializationCountingValue sc = new SerializationCountingValue();
      // Update a key upon which the client has expressed interest,
      // expect it to send an update message to the client
      rootRegion.put(ks2, sc);

      // Wait to know that the data has been at least serialized (possibly sent)
      await()
          .until(() -> ccp.getStatistics().getMessagesProcessed() != currMesgCount);

      // assert one serialization to send value to interested client
      // more than one implies copy-on-read behavior (bad)
      assertThat(sc.count.get()).isEqualTo(1);
      assertThat(ks2.count.get()).isEqualTo(1);
    });

    // Double-check the serialization count in the event that the previous check
    // missed the copy due to race conditions
    client.invoke("Assert copy behavior from client after update", () -> {
      Region rootRegion = getRootRegion(rName);
      { // Once to send the value to this client via the updater thread

        SerializationCountingKey ks2 = new SerializationCountingKey(k2);
        // Wait for the update to arrive on to the Cache Client Updater
        final int maxSecs = 30;
        await("Waited over " + maxSecs + "s").timeout(maxSecs, TimeUnit.SECONDS)
            .until(() -> rootRegion.containsKey(ks2));

        SerializationCountingValue sc =
            (SerializationCountingValue) rootRegion.getEntry(ks2).getValue();
        assertThat(sc.count.get()).isEqualTo(1);
      }
    });
  }
Test that CopyOnRead doesn't cause {@link HARegionQueue#peek()} to create a copy, assuming that
creating copies performs a serialize and de-serialize operation.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/ClientDeserializationCopyOnReadRegressionTest.java