@Test
  public void test015InvalidateAndDestroyToEmptyAllPropagation() throws CacheException {
    final String name = getName();

    vm0.invoke("Create Cache Server", () -> {
      RegionFactory<Object, Object> factory = getBridgeServerRegionAttributes(null, null);
      createRegion(name, factory);
      startBridgeServer(0);
    });
    final int port = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName();

    vm1.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      configureConnectionPool(factory, host0, new int[] {port}, true, -1, -1, null);
      factory.addCacheListener(new CertifiableTestCacheListener<>());
      factory.setDataPolicy(DataPolicy.EMPTY);
      factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
      Region rgn = createRegion(name, factory);
      rgn.registerInterestRegex(".*", false, false);
    });
    vm1.invoke("Populate region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        region.put(i, "old" + i);
      }
    });

    vm2.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      configureConnectionPool(factory, host0, new int[] {port}, true, -1, -1, null);
      factory.addCacheListener(new CertifiableTestCacheListener<>());
      Region rgn = createRegion(name, factory);
      rgn.registerInterestRegex(".*", false, false);
    });

    vm1.invoke("Turn on history", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      CertifiableTestCacheListener<Object, Object> ctl =
          (CertifiableTestCacheListener<Object, Object>) region.getAttributes()
              .getCacheListeners()[0];
      ctl.enableEventHistory();
    });

    vm2.invoke("Update region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        region.put(i, "new" + i, "callbackArg" + i);
      }
    });

    vm1.invoke("Verify invalidates", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      CertifiableTestCacheListener<Object, Object> ctl =
          (CertifiableTestCacheListener<Object, Object>) region.getAttributes()
              .getCacheListeners()[0];

      for (int i = 0; i < 10; i++) {
        Object key = i;
        ctl.waitForInvalidated(key);
        await().until(() -> region.getEntry(key) == null);
      }
      {
        List<CacheEvent<Object, Object>> list = ctl.getEventHistory();
        assertThat(list.size()).isEqualTo(10);
        for (int i = 0; i < 10; i++) {
          Object key = i;
          EntryEvent ee = (EntryEvent) list.get(i);
          assertThat(ee.getKey()).isEqualTo(key);
          assertThat(ee.getOldValue()).isNull();
          assertThat(ee.isOldValueAvailable()).isFalse(); // failure
          assertThat(Operation.INVALIDATE).isEqualTo(ee.getOperation());
          assertThat("callbackArg" + i).isEqualTo(ee.getCallbackArgument());
          assertThat(ee.isOriginRemote()).isTrue();
        }
      }
    });

    vm2.invoke("Validate original and destroy", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        Object key = i;
        assertThat("new" + i).isEqualTo(region.getEntry(key).getValue());
        region.destroy(key, "destroyCB" + i);
      }
    });

    vm1.invoke("Verify destroys", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      CertifiableTestCacheListener<Object, Object> ctl =
          (CertifiableTestCacheListener<Object, Object>) region.getAttributes()
              .getCacheListeners()[0];
      for (int i = 0; i < 10; i++) {
        Object key = i;
        ctl.waitForDestroyed(key);
        await().until(() -> region.getEntry(key) == null);
      }
      {
        List<CacheEvent<Object, Object>> list = ctl.getEventHistory();
        assertThat(10).isEqualTo(list.size());
        for (int i = 0; i < 10; i++) {
          Object key = i;
          EntryEvent ee = (EntryEvent) list.get(i);
          assertThat(ee.getKey()).isEqualTo(key);
          assertThat(ee.getOldValue()).isNull();
          assertThat(ee.isOldValueAvailable()).isFalse();
          assertThat(Operation.DESTROY).isEqualTo(ee.getOperation());
          assertThat("destroyCB" + i).isEqualTo(ee.getCallbackArgument());
          assertThat(ee.isOriginRemote()).isTrue();
        }
      }
    });
    vm2.invoke("recreate", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        Object key = i;
        region.create(key, "create" + i, "createCB" + i);
      }
    });

    vm1.invoke("Verify creates", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      CertifiableTestCacheListener<Object, Object> ctl =
          (CertifiableTestCacheListener<Object, Object>) region.getAttributes()
              .getCacheListeners()[0];
      for (int i = 0; i < 10; i++) {
        Object key = i;
        ctl.waitForInvalidated(key);
        await().until(() -> region.getEntry(key) == null);
      }
      List<CacheEvent<Object, Object>> list = ctl.getEventHistory();
      assertThat(10).isEqualTo(list.size());
      for (int i = 0; i < 10; i++) {
        Object key = i;
        EntryEvent ee = (EntryEvent) list.get(i);
        assertThat(ee.getKey()).isEqualTo(key);
        assertThat(ee.getOldValue()).isNull();
        assertThat(ee.isOldValueAvailable()).isFalse();
        assertThat(Operation.INVALIDATE).isEqualTo(ee.getOperation());
        assertThat("createCB" + i).isEqualTo(ee.getCallbackArgument());
        assertThat(ee.isOriginRemote()).isTrue();
      }
      // now see if we can get it from the server
      for (int i = 0; i < 10; i++) {
        Object key = i;
        assertThat("create" + i).isEqualTo(region.get(key, "loadCB" + i));
      }

      list = ctl.getEventHistory();
      assertThat(10).isEqualTo(list.size());
      for (int i = 0; i < 10; i++) {
        Object key = i;
        EntryEvent ee = (EntryEvent) list.get(i);
        assertThat(ee.getKey()).isEqualTo(key);
        assertThat(ee.getOldValue()).isNull();
        assertThat(ee.getNewValue()).isEqualTo("create" + i);
        assertThat(Operation.LOCAL_LOAD_CREATE).isEqualTo(ee.getOperation());
        assertThat("loadCB" + i).isEqualTo(ee.getCallbackArgument());
        assertThat(ee.isOriginRemote()).isFalse();
      }
    });

    Stream.of(vm1, vm2).forEach(vm -> vm.invoke("Close Pool", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.localDestroyRegion();
    }));


  }
Tests that invalidates and destroys are propagated to {@link Pool}s correctly to
DataPolicy.EMPTY + InterestPolicy.ALL

@since GemFire 5.0
geode-core/src/distributedTest/java/org/apache/geode/cache/ConnectionPoolDUnitTest.java