@Test
  public void testPartitionedRegionAndCopyOnRead() {
    final VM accessor = VM.getVM(2);
    final VM datastore = VM.getVM(3);
    final String rName = getUniqueName();
    final String k1 = "k1";

    datastore.invoke("Create PR DataStore", () -> {
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setPartitionAttributes(
          new PartitionAttributesFactory().setRedundantCopies(0).create());
      createRootRegion(rName, factory);
    });

    accessor.invoke("Create PR Accessor and put new value", () -> {
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setPartitionAttributes(
          new PartitionAttributesFactory().setLocalMaxMemory(0).setRedundantCopies(0).create());
      Region<Object, Object> rootRegion = createRootRegion(rName, factory);
      SerializationCountingValue val = new SerializationCountingValue();
      rootRegion.put(k1, val);
      // First put to a bucket will serialize once to determine the size of the value
      // to know how much extra space the new bucket with the new entry will consume
      // and serialize again to send the bytes
      assertThat(val.count.get()).isEqualTo(2);
      // A put to an already created bucket should only be serialized once
      val = new SerializationCountingValue();
      rootRegion.put(k1, val);
      assertThat(val.count.get()).isEqualTo(1);
    });

    datastore.invoke("assert datastore entry serialization count", () -> {
      PartitionedRegion pr = (PartitionedRegion) getRootRegion(rName);
      // Visit the one bucket (since there is only one value in the entire PR)
      // to directly copy the entry bytes and assert the serialization count.
      // All this extra work is to assure the serialization count does not increase
      // (by de-serializing the value stored in the map, which would then have to be
      // re-serialized).
      pr.getDataStore().visitBuckets((bucketId, r) -> {
        BucketRegion br = (BucketRegion) r;
        KeyInfo keyInfo = new KeyInfo(k1, null, bucketId);
        RawValue rv = null;
        try {
          rv = br.getSerialized(keyInfo, false, false, null, null, false);
        } catch (IOException e) {
          fail("Unexpected IOException", e);
        }
        Object val = rv.getRawValue();
        assertThat(val).isInstanceOf(CachedDeserializable.class);
        CachedDeserializable cd = (CachedDeserializable) val;
        SerializationCountingValue scv =
            (SerializationCountingValue) cd.getDeserializedForReading();
        assertThat(scv.count.get()).isEqualTo(1);
      });
    });

    accessor.invoke("assert accessor entry serialization count", () -> {
      Region<Object, Object> rootRegion = getRootRegion(rName);
      SerializationCountingValue value1 = (SerializationCountingValue) rootRegion.get(k1);
      // The counter was incremented once to send the data to the datastore
      assertThat(value1.count.get()).isEqualTo(1);
      getCache().setCopyOnRead(true);
      // Once to send the data to the datastore, no need to do a serialization
      // when we make copy since it is serialized from datastore to us.
      SerializationCountingValue value2 = (SerializationCountingValue) rootRegion.get(k1);
      assertThat(value2.count.get()).isEqualTo(1);
      assertThat(value2).isNotEqualTo(value1);
    });

    datastore.invoke("assert value serialization", () -> {
      Region rootRegion = getRootRegion(rName);
      SerializationCountingValue value1 = (SerializationCountingValue) rootRegion.get(k1);
      // Once to send the value from the accessor to the data store
      assertThat(value1.count.get()).isEqualTo(1);
      getCache().setCopyOnRead(true);
      // Once to send the value from the accessor to the data store
      // once to make a local copy
      SerializationCountingValue value2 = (SerializationCountingValue) rootRegion.get(k1);
      assertThat(value2.count.get()).isEqualTo(2);
      assertThat(value2).isNotEqualTo(value1);
    });
  }
Test to ensure that a PartitionedRegion doesn't make more than the expected number of copies
when copy-on-read is set to true
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/ClientDeserializationCopyOnReadRegressionTest.java