@Test
  public void testNonblockingGetInitialImage() throws InterruptedException {
    assumeThat(supportsReplication()).isTrue();
    // don't run this test if global scope since its too difficult to predict
    // how many concurrent operations will occur
    assumeThat(getRegionAttributes().getScope().isGlobal()).isFalse();

    final String name = getUniqueName();
    final byte[][] values = new byte[NB1_NUM_ENTRIES][];

    for (int i = 0; i < NB1_NUM_ENTRIES; i++) {
      values[i] = new byte[NB1_VALUE_SIZE];
      Arrays.fill(values[i], (byte) 0x42);
    }

    vm0.invoke("Create Nonmirrored Region", () -> {
      { // root region must be DACK because its used to sync up async subregions
        AttributesFactory<?, ?> factory = new AttributesFactory<>();
        factory.setScope(Scope.DISTRIBUTED_ACK);
        factory.setDataPolicy(DataPolicy.EMPTY);
        createRootRegion(factory.create());
      }
      {
        AttributesFactory<?, ?> factory = new AttributesFactory<>(getRegionAttributes());
        createRegion(name, factory.create());
      }
      // reset slow
      InitialImageOperation.slowImageProcessing = 0;
    });

    vm0.invoke("Put initial data", () -> {
      Region<Integer, byte[]> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < NB1_NUM_ENTRIES; i++) {
        region.put(i, values[i]);
      }
      assertThat(region.keySet().size()).isEqualTo(NB1_NUM_ENTRIES);
    });

    // start asynchronous process that does updates to the data
    AsyncInvocation<Void> async =
        vm0.invokeAsync("Do Nonblocking Operations", () -> {
          Region<Object, Object> region = getRootRegion().getSubregion(name);

          // wait for profile of getInitialImage cache to show up
          final org.apache.geode.internal.cache.CacheDistributionAdvisor adv =
              ((org.apache.geode.internal.cache.DistributedRegion) region)
                  .getCacheDistributionAdvisor();
          final int expectedProfiles = 1;

          await("replicate count never reached " + expectedProfiles)
              .until(() -> {
                DataPolicy currentPolicy = getRegionAttributes().getDataPolicy();
                if (currentPolicy == DataPolicy.PRELOADED) {
                  return (adv.advisePreloadeds().size()
                      + adv.adviseReplicates().size()) >= expectedProfiles;
                } else {
                  return adv.adviseReplicates().size() >= expectedProfiles;
                }
              });

          DataPolicy currentPolicy = getRegionAttributes().getDataPolicy();
          int numProfiles;
          if (currentPolicy == DataPolicy.PRELOADED) {
            numProfiles = adv.advisePreloadeds().size() + adv.adviseReplicates().size();
          } else {
            numProfiles = adv.adviseReplicates().size();
          }
          assertThat(numProfiles >= expectedProfiles).isTrue();

          // operate on every odd entry with different value, alternating between
          // updates, invalidates, and destroys. These operations are likely
          // to be nonblocking if a sufficient number of updates get through
          // before the get initial image is complete.
          for (int i = 1; i < NB1_NUM_ENTRIES; i += 2) {
            Object key = i;
            logger.info("Operation #" + i + " on key " + key);
            switch (i % 6) {
              case 1: // UPDATE
                // use the current timestamp so we know when it happened
                // we could have used last modification timestamps, but
                // this works without enabling statistics
                Object value = System.currentTimeMillis();
                region.put(key, value);
                // no longer safe since get is not allowed to member doing GII
                // if (getRegionAttributes().getScope().isDistributedAck()) {
                // // do a nonblocking netSearch
                // region.localInvalidate(key);
                // assertIndexDetailsEquals(value, region.get(key));
                // }
                break;
              case 3: // INVALIDATE
                region.invalidate(key);
                if (getRegionAttributes().getScope().isDistributedAck()) {
                  // do a nonblocking netSearch
                  assertThat(region.get(key)).isNull();
                }
                break;
              case 5: // DESTROY
                region.destroy(key);
                if (getRegionAttributes().getScope().isDistributedAck()) {
                  // do a nonblocking netSearch
                  assertThat(region.get(key)).isNull();
                }
                break;
              default:
                fail("unexpected modulus result: " + i);
                break;
            }
          }

          // add some new keys
          for (int i = NB1_NUM_ENTRIES; i < NB1_NUM_ENTRIES + 200; i++) {
            region.create(i, System.currentTimeMillis());
          }
          // now do a put and our DACK root region which will not complete
          // until processed on otherside which means everything done before this
          // point has been processed
          getRootRegion().put("DONE", "FLUSH_OPS");
        });

    // in the meantime, do the get initial image in vm2
    // slow down image processing to make it more likely to get async updates
    if (!getRegionAttributes().getScope().isGlobal()) {
      vm2.invoke("Set slow image processing", () -> {
        // if this is a no_ack test, then we need to slow down more because of the
        // pauses in the nonblocking operations
        InitialImageOperation.slowImageProcessing = 200;
      });
    }

    SerializableRunnable create = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        beginCacheXml();
        { // root region must be DACK because its used to sync up async subregions
          AttributesFactory<?, ?> factory = new AttributesFactory<>();
          factory.setScope(Scope.DISTRIBUTED_ACK);
          factory.setDataPolicy(DataPolicy.NORMAL);
          factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
          createRootRegion(factory.create());
        }
        {
          AttributesFactory<?, ?> factory = new AttributesFactory<>(getRegionAttributes());
          factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
          if (getRegionAttributes().getDataPolicy() == DataPolicy.NORMAL) {
            factory.setDataPolicy(DataPolicy.PRELOADED);
          }
          createRegion(name, factory.create());
        }
        finishCacheXml(name);
        // reset slow
        org.apache.geode.internal.cache.InitialImageOperation.slowImageProcessing = 0;
      }
    };

    AsyncInvocation<Void> asyncGII = vm2.invokeAsync("Create Mirrored Region", create);

    if (!getRegionAttributes().getScope().isGlobal()) {
      // wait for nonblocking operations to complete
      async.get();
      vm2.invoke("Set fast image processing", () -> {
        InitialImageOperation.slowImageProcessing = 0;
      });
    }

    // wait for GII to complete
    asyncGII.get();
    final long iiComplete = System.currentTimeMillis();

    if (getRegionAttributes().getScope().isGlobal()) {
      // wait for nonblocking operations to complete
      async.get();
    }

    // Locally destroy the region in vm0 so we know that they are not found by
    // a netSearch
    vm0.invoke("Locally destroy region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.localDestroyRegion();
    });

    // invoke repeating so noack regions wait for all updates to get processed
    vm2.invoke("Verify entryCount", repeatingIfNecessary(5000, () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      // expected entry count (subtract entries destroyed)
      int entryCount = NB1_NUM_ENTRIES + 200 - NB1_NUM_ENTRIES / 6;
      int actualCount = region.entrySet(false).size();

      assertThat(actualCount).isEqualTo(entryCount);
    }));

    vm2.invoke("Verify keys/values & Nonblocking", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      // expected entry count (subtract entries destroyed)
      int entryCount = NB1_NUM_ENTRIES + 200 - NB1_NUM_ENTRIES / 6;
      assertThat(region.entrySet(false).size()).isEqualTo(entryCount);
      // determine how many entries were updated before getInitialImage
      // was complete
      int numConcurrent = 0;
      numConcurrent = verifyEntryUpdateCounts(region, numConcurrent, values, iiComplete);

      // make sure at least some of them were concurrent
      if (region.getAttributes().getScope().isGlobal()) {
        assertThat(numConcurrent < 10).describedAs(
            "Too many concurrent updates when expected to block: " + numConcurrent).isTrue();
      } else {
        int min = 30;
        assertThat(numConcurrent >= min).describedAs(
            "Not enough updates concurrent with getInitialImage occurred to my liking. "
                + numConcurrent + " entries out of " + entryCount
                + " were updated concurrently with getInitialImage, and I'd expect at least "
                + min
                + " or so")
            .isTrue();
      }
    });
  }
Tests to makes sure that a distributed update resets the expiration timer.
geode-dunit/src/main/java/org/apache/geode/cache30/MultiVMRegionTestCase.java