@Test
  public void testConcurrentOpWithGII() throws InterruptedException {
    if (getClass() != DistributedAckRegionCCEDUnitTest.class) {
      return; // not really a scope-related thing
    }
    final String name = getUniqueName() + "-CC";
    final String key = "mykey";
    VM vm1 = VM.getVM(1);
    VM vm2 = VM.getVM(2);

    // create some destroyed entries so the GC service is populated
    // do conflicting update() and destroy() on the region. We want the update() to
    // be sent with a message and the destroy() to be transferred in the initial image
    // and be the value that we want to keep
    vm1.invoke("create region", () -> {
      RegionFactory f1 = getCache().createRegionFactory(getRegionAttributes());
      CCRegion = (LocalRegion) f1.create(name);
      return CCRegion.getDistributionManager().getDistributionManagerId();
    });

    AsyncInvocation<Void> partialCreate =
        vm2.invokeAsync("create region with stall", () -> {

          final GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
          RegionFactory f = cache.createRegionFactory(getRegionAttributes());
          InitialImageOperation.VMOTION_DURING_GII = true;
          // this will stall region creation at the point of asking for an initial image
          VMotionObserverHolder.setInstance(new VMotionObserver() {
            @Override
            public void vMotionBeforeCQRegistration() {}

            @Override
            public void vMotionBeforeRegisterInterest() {}

            @Override
            public void vMotionDuringGII(Set recipientSet, LocalRegion region) {
              InitialImageOperation.VMOTION_DURING_GII = false;
              final InitializationLevel oldLevel =
                  LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);
              LocalRegion ccregion = (LocalRegion) cache.getInternalRegionByPath(SEPARATOR + name);
              try {
                // wait for the update op (sent below from vm1) to arrive, then allow the GII to
                // happen
                await().ignoreException(RegionDestroyedException.class)
                    .until(() -> ccregion.isDestroyed() || ccregion.getRegionEntry(key) != null);
              } finally {
                LocalRegion.setThreadInitLevelRequirement(oldLevel);
              }
            }
          });
          try {
            CCRegion = (LocalRegion) f.create(name);
            // at this point we should have received the update op and then the GII, which should
            // overwrite the conflicting update op
            assertThat(CCRegion.containsKey(key))
                .describedAs("entry was destroyed by initial image transfer")
                .isFalse();
          } finally {
            InitialImageOperation.VMOTION_DURING_GII = false;
          }
          return null;
        });

    vm1.invoke("create conflicting events", () -> {
      // wait for the other to come on line
      DistributionAdvisor adv = ((DistributedRegion) CCRegion).getCacheDistributionAdvisor();

      await().until(() -> !adv.adviseGeneric().isEmpty());
      assertThat(adv.adviseGeneric())
          .withFailMessage("other member never came on line")
          .isNotEmpty();

      DistributedCacheOperation.LOSS_SIMULATION_RATIO = 200.0; // inhibit all messaging
      try {
        CCRegion.put("mykey", "initialValue");
        CCRegion.destroy("mykey");
      } finally {
        DistributedCacheOperation.LOSS_SIMULATION_RATIO = 0.0;
      }

      // generate a fake version tag for the message
      final VersionStamp versionStamp = CCRegion.getRegionEntry(key).getVersionStamp();
      VersionTag<InternalDistributedMember> tag =
          (VersionTag<InternalDistributedMember>) versionStamp.asVersionTag();
      // create a fake member ID that will be < mine and lose a concurrency check
      InternalDistributedMember nm =
          CCRegion.getDistributionManager().getDistributionManagerId();
      InternalDistributedMember mbr = null;

      mbr = new InternalDistributedMember(nm.getInetAddress().getCanonicalHostName(),
          nm.getMembershipPort() - 1, "fake_id", "fake_id_ustring",
          ClusterDistributionManager.NORMAL_DM_TYPE, null, null);
      tag.setMemberID(mbr);


      // generate an event to distribute that contains the fake version tag
      EntryEventImpl event =
          EntryEventImpl.create(CCRegion, Operation.UPDATE, key, false, mbr, true, false);
      event.setNewValue("newValue");
      event.setVersionTag(tag);

      // this should update the controller's cache with the updated value but leave this cache
      // alone
      DistributedCacheOperation op = new UpdateOperation(event, tag.getVersionTimeStamp());
      op.distribute();
      event.release();
    });

    partialCreate.get();

  }
test for bug #45564. a create() is received by region creator and then a later destroy() is
received in initial image and while the version info from the destroy is recorded we keep the
value from the create event
geode-core/src/distributedTest/java/org/apache/geode/cache30/DistributedAckRegionCCEDUnitTest.java