@Test
  public void testRemoteCacheListener() {
    assertThat(getRegionAttributes().getScope().isDistributed()).isTrue();

    final String name = getUniqueName();
    final Object key = "KEY";
    final Object oldValue = "OLD_VALUE";
    final Object newValue = "NEW_VALUE";

    vm0.invoke("Create Region and Put", () -> {
      Region<Object, Object> region1 = createRegion(name);
      region1.put(key, oldValue);
    });
    vm1.invoke("Create Region and Put", () -> {
      Region<Object, Object> region1 = createRegion(name);
      region1.put(key, oldValue);
    });

    vm1.invoke("Set listener", () -> {
      final Region<Object, Object> region = getRootRegion().getSubregion(name);
      setListener(new TestCacheListener<Object, Object>() {
        @Override
        public void afterUpdate2(EntryEvent<Object, Object> event) {
          assertThat(event.getOperation()).isEqualTo(Operation.UPDATE);
          assertThat(event.getRegion()).isEqualTo(region);
          assertThat(event.getOperation().isDistributed()).isTrue();
          assertThat(event.getOperation().isExpiration()).isFalse();
          assertThat(event.isOriginRemote()).isTrue();
          assertThat(event.getDistributedMember()).isEqualTo(event.getCallbackArgument());
          assertThat(event.getKey()).isEqualTo(key);
          assertThat(event.getOldValue()).isEqualTo(oldValue);
          assertThat(event.getNewValue()).isEqualTo(newValue);
          assertThat(event.getOperation().isLoad()).isFalse();
          assertThat(event.getOperation().isLocalLoad()).isFalse();
          assertThat(event.getOperation().isNetLoad()).isFalse();
          assertThat(event.getOperation().isNetSearch()).isFalse();
          if (event.getRegion().getAttributes().getOffHeap()) {
            // since off heap always serializes the old value is serialized and available
            assertThat(event.getSerializedOldValue().getDeserializedValue()).isEqualTo(oldValue);
          } else {
            assertThat(event.getSerializedOldValue()).isNull(); // since it was put originally in
            // this VM
          }
          DataInputStream dis = new DataInputStream(
              new ByteArrayInputStream(event.getSerializedNewValue().getSerializedValue()));
          try {
            assertThat(DataSerializer.<Object>readObject(dis)).isEqualTo(newValue);
          } catch (IOException | ClassNotFoundException e) {
            fail("Unexpected Exception", e);
          }
        }
      });
      region.getAttributesMutator().addCacheListener(listener());
    });

    // I see no reason to pause here.
    // The test used to pause here but only if no-ack.
    // But we have no operations to wait for.
    // The last thing we did was install a listener in vm1
    // and it is possible that vm0 does not yet know we have
    // a listener but for this test it does not matter.
    // So I'm commenting out the following pause:
    // pauseIfNecessary();
    // If needed then do a flushIfNecessary(region) after adding the cache listener

    vm0.invoke("Update", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.put(key, newValue, getSystem().getDistributedMember());
    });

    vm1.invoke("Verify Update", () -> {
      listener().waitForInvocation(3000, 10);

      // Setup listener for next test
      final Region<Object, Object> region = getRootRegion().getSubregion(name);
      setListener(new TestCacheListener<Object, Object>() {
        @Override
        public void afterInvalidate2(EntryEvent<Object, Object> event) {
          assertThat(event.getOperation()).isEqualTo(Operation.INVALIDATE);
          assertThat(event.getRegion()).isEqualTo(region);
          assertThat(event.getOperation().isDistributed()).isTrue();
          assertThat(event.getOperation().isExpiration()).isFalse();
          assertThat(event.isOriginRemote()).isTrue();
          assertThat(event.getDistributedMember()).isEqualTo(event.getCallbackArgument());
          assertThat(event.getKey()).isEqualTo(key);
          assertThat(event.getOldValue()).isEqualTo(newValue);
          assertThat(event.getNewValue()).isNull();
          assertThat(event.getOperation().isLoad()).isFalse();
          assertThat(event.getOperation().isLocalLoad()).isFalse();
          assertThat(event.getOperation().isNetLoad()).isFalse();
          assertThat(event.getOperation().isNetSearch()).isFalse();
          assertThat(event.getSerializedNewValue()).isNull();
          DataInputStream dis = new DataInputStream(
              new ByteArrayInputStream(event.getSerializedOldValue().getSerializedValue()));
          try {
            assertThat(DataSerializer.<Object>readObject(dis)).isEqualTo(newValue);
          } catch (IOException | ClassNotFoundException e) {
            fail("Unexpected Exception", e);
          }
        }
      });
      region.getAttributesMutator().addCacheListener(listener());
    });

    vm0.invoke("Invalidate", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.invalidate(key, getSystem().getDistributedMember());
    });

    vm1.invoke("Verify Invalidate", () -> {
      listener().waitForInvocation(3000, 10);

      // Setup listener for next test
      final Region<Object, Object> region = getRootRegion().getSubregion(name);
      setListener(new TestCacheListener<Object, Object>() {
        @Override
        public void afterDestroy2(EntryEvent<Object, Object> event) {
          assertThat(event.getOperation().isDestroy()).isTrue();
          assertThat(event.getRegion()).isEqualTo(region);
          assertThat(event.getOperation().isDistributed()).isTrue();
          assertThat(event.getOperation().isExpiration()).isFalse();
          assertThat(event.isOriginRemote()).isTrue();
          assertThat(event.getDistributedMember()).isEqualTo(event.getCallbackArgument());
          assertThat(event.getKey()).isEqualTo(key);
          assertThat(event.getOldValue()).isNull();
          assertThat(event.getNewValue()).isNull();
          assertThat(event.getOperation().isLoad()).isFalse();
          assertThat(event.getOperation().isLocalLoad()).isFalse();
          assertThat(event.getOperation().isNetLoad()).isFalse();
          assertThat(event.getOperation().isNetSearch()).isFalse();
          assertThat(event.getSerializedOldValue()).isNull();
          assertThat(event.getSerializedNewValue()).isNull();
        }
      });
      region.getAttributesMutator().addCacheListener(listener());
    });

    vm0.invoke("Destroy", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.destroy(key, getSystem().getDistributedMember());
    });

    vm1.invoke("Verify Destroy", () -> {
      listener().waitForInvocation(3000, 10);

      // Setup listener for next test
      final Region<Object, Object> region = getRootRegion().getSubregion(name);
      setListener(new TestCacheListener<Object, Object>() {
        @Override
        public void afterRegionInvalidate2(RegionEvent<Object, Object> event) {
          assertThat(event.getOperation()).isEqualTo(Operation.REGION_INVALIDATE);
          assertThat(event.getRegion()).isEqualTo(region);
          assertThat(event.getOperation().isDistributed()).isTrue();
          assertThat(event.getOperation().isExpiration()).isFalse();
          assertThat(event.isOriginRemote()).isTrue();
          assertThat(event.getDistributedMember()).isEqualTo(event.getCallbackArgument());
        }
      });
      region.getAttributesMutator().addCacheListener(listener());
    });

    vm0.invoke("Invalidate Region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.invalidateRegion(getSystem().getDistributedMember());
    });

    vm1.invoke("Verify Invalidate Region", () -> {
      listener().waitForInvocation(3000, 10);

      // Setup listener for next test
      final Region<Object, Object> region = getRootRegion().getSubregion(name);
      setListener(new TestCacheListener<Object, Object>() {
        @Override
        public void afterRegionDestroy2(RegionEvent<Object, Object> event) {
          assertThat(event.getOperation()).isEqualTo(Operation.REGION_DESTROY);
          assertThat(event.getRegion()).isEqualTo(region);
          assertThat(event.getOperation().isDistributed()).isTrue();
          assertThat(event.getOperation().isExpiration()).isFalse();
          assertThat(event.isOriginRemote()).isTrue();
          assertThat(event.getDistributedMember()).isEqualTo(event.getCallbackArgument());
        }
      });
      region.getAttributesMutator().addCacheListener(listener());
    });

    vm0.invoke("Destroy Region", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.destroyRegion(getSystem().getDistributedMember());
    });

    vm1.invoke("Verify Destroy Region", () -> {
      listener().waitForInvocation(3000, 10);
    });
  }
Tests that a {@link CacheListener} is invoked in a remote VM.
geode-dunit/src/main/java/org/apache/geode/cache30/MultiVMRegionTestCase.java