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

    final String name = getUniqueName();

    vm0.invoke("Create Region", () -> {
      createRegion(name);
    });
    vm1.invoke("Create Region", () -> {
      createRegion(name);
    });

    final Object key = "KEY";
    final Object key2 = "KEY2";
    final int intValue = 3452;
    final long longValue = 213421;
    // final boolean[] wasInvoked = new boolean[1];

    vm2.invoke("Disconnect from DS", JUnit4DistributedTestCase::disconnectFromDS);

    try {
      vm0.invoke("Put int", () -> {
        Class c = IntWrapper.IntWrapperSerializer.class;
        IntWrapper.IntWrapperSerializer serializer =
            (IntWrapper.IntWrapperSerializer) DataSerializer.register(c);

        Region<Object, Object> region = getRootRegion().getSubregion(name);
        region.put(key, new IntWrapper(intValue));

        flushIfNecessary(region);
        assertThat(serializer.wasInvoked).isTrue();
      });

      SerializableRunnable get = new CacheSerializableRunnable() {
        @Override
        public void run2() throws CacheException {
          Region<Object, Object> region = getRootRegion().getSubregion(name);
          // wait a while for the serializer to be registered
          // A race condition exists in the product in which
          // this thread can be stuck waiting in getSerializer
          // for 60 seconds. So it only calls getSerializer once
          // causing it to fail intermittently (see GEODE-376).
          // To workaround this the test wets WAIT_MS to 1 ms.
          // So the getSerializer will only block for 1 ms.
          // This allows the WaitCriterion to make multiple calls
          // of getSerializer and the subsequent calls will find
          // the DataSerializer.
          final int savVal = InternalDataSerializer.GetMarker.WAIT_MS;
          InternalDataSerializer.GetMarker.WAIT_MS = 1;
          try {
            await("DataSerializer with id 120 was never registered")
                .until(() -> InternalDataSerializer.getSerializer((byte) 120) != null);
          } finally {
            InternalDataSerializer.GetMarker.WAIT_MS = savVal;
          }
          IntWrapper value = (IntWrapper) region.get(key);
          assertThat(InternalDataSerializer.getSerializer((byte) 120)).isNotNull();
          assertThat(value).isNotNull();
          assertThat(value.intValue).isEqualTo(intValue);
        }
      };
      vm1.invoke("Get int", get);

      // Make sure that VMs that connect after registration can get the
      // serializer
      vm2.invoke("Connect to DS", () -> {
        // Register a DataSerializer before connecting to system
        Class c = LongWrapper.LongWrapperSerializer.class;
        DataSerializer.register(c);

        getSystem();
      });
      vm2.invoke("Create Region", () -> {
        createRegion(name);
      });

      // Make sure vm0 and vm1 have both registered the LongWrapperSerializer after vm2 reconnects
      // and before vm2 puts any instance of LongWrapper in the region
      // This is to avoid async disk store FlusherThread serializing any instance of LongWrapper
      // before the LongWrapperSerializer is registered
      vm0.invoke("waitForLongWrapperSerializerRegistration",
          this::waitForLongWrapperSerializerRegistration);

      // see the comments in "get" CacheSerializableRunnable above
      vm1.invoke("waitForLongWrapperSerializerRegistration",
          this::waitForLongWrapperSerializerRegistration);

      vm2.invoke("Put long", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        region.put(key2, new LongWrapper(longValue));

        flushIfNecessary(region);

        LongWrapper.LongWrapperSerializer serializer =
            (LongWrapper.LongWrapperSerializer) InternalDataSerializer.getSerializer((byte) 121);
        assertThat(serializer.wasInvoked).isTrue();
      });
      vm2.invoke("Get int", get);

      SerializableRunnable get2 = new CacheSerializableRunnable() {
        @Override
        public void run2() throws CacheException {
          Region<Object, Object> region = getRootRegion().getSubregion(name);
          LongWrapper value = (LongWrapper) region.get(key2);
          assertThat(InternalDataSerializer.getSerializer((byte) 121)).isNotNull();
          assertThat(value).isNotNull();
          assertThat(value.longValue).isEqualTo(longValue);
        }
      };
      vm0.invoke("Get long", get2);
      vm1.invoke("Get long", get2);

      // wait a little while for other netsearch requests to return
      // before unregistering the serializers that will be needed to process these
      // responses.
    } finally {
      Wait.pause(1500);
      Stream.of(vm0, vm1, vm2).forEach(vm -> vm.invoke("Flush disk store",
          () -> flushDiskStore(name)));
      unregisterAllSerializers();
    }
  }
Tests what happens when one VM attempts to read an object for which it does not have a
registered <code>DataSerializer</code>.

@since GemFire 3.5
geode-dunit/src/main/java/org/apache/geode/cache30/MultiVMRegionTestCase.java