@Test
  public void testWriteCorrectVersionToKrf() throws Exception {
    VM vm1 = getVM(1);

    InternalRegion region = (InternalRegion) createAsyncRegionWithSmallQueue(vm1);

    // The idea here is to do a bunch of puts with async persistence
    // At some point the oplog will switch. At that time, we wait for a krf
    // to be created and then throw an exception to shutdown the disk store.
    //
    // This should cause us to create a krf with some entries that have been
    // modified since the crf was written and are still in the async queue.
    //
    // To avoid deadlocks, we need to mark that the oplog was switched,
    // and then do the wait in the flusher thread.

    // Setup the callbacks to wait for krf creation and throw an exception
    LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true;
    try (IgnoredException e = addIgnoredException(DiskAccessException.class)) {
      CountDownLatch krfCreated = new CountDownLatch(1);
      AtomicBoolean oplogSwitched = new AtomicBoolean(false);

      CacheObserverHolder.setInstance(new CacheObserverAdapter() {

        @Override
        public void afterKrfCreated() {
          krfCreated.countDown();
        }

        @Override
        public void afterWritingBytes() {
          if (oplogSwitched.get()) {
            errorCollector
                .checkSucceeds(() -> assertThat(krfCreated.await(3_000, SECONDS)).isTrue());

            throw new DiskAccessException();
          }
        }

        @Override
        public void afterSwitchingOplog() {
          oplogSwitched.set(true);
        }
      });

      // This is just to make sure the first oplog is not completely garbage.
      region.put("testkey", "key");

      // Do some puts to trigger an oplog roll.
      try {
        // Starting with a value of 1 means the value should match
        // the region version for easier debugging.
        int i = 1;
        while (krfCreated.getCount() > 0) {
          i++;
          region.put("key" + (i % 3), i);
          sleep(2);
        }
      } catch (CacheClosedException | DiskAccessException expected) {
        // do nothing
      }

      // Wait for the region to be destroyed. The region won't be destroyed
      // until the async flusher thread ends up switching oplogs
      await().until(region::isDestroyed);
      closeCache();
    } finally {
      LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = false;
      CacheObserverHolder.setInstance(null);
    }

    // Get the version tags from the krf
    InternalRegion recoveredRegion = (InternalRegion) createAsyncRegionWithSmallQueue(vm1);
    VersionTag[] tagsFromKrf = new VersionTag[3];
    for (int i = 0; i < 3; i++) {
      NonTXEntry entry = (NonTXEntry) recoveredRegion.getEntry("key" + i);
      tagsFromKrf[i] = entry.getRegionEntry().getVersionStamp().asVersionTag();
      LogWriterUtils.getLogWriter()
          .info("krfTag[" + i + "]=" + tagsFromKrf[i] + ",value=" + entry.getValue());
    }

    closeCache();

    // Set a system property to skip recovering from the krf so we can
    // get the version tag from the crf.
    System.setProperty(DiskStoreImpl.RECOVER_VALUES_SYNC_PROPERTY_NAME, "true");

    // Get the version tags from the crf
    recoveredRegion = (InternalRegion) createAsyncRegionWithSmallQueue(vm1);
    VersionTag[] tagsFromCrf = new VersionTag[3];
    for (int i = 0; i < 3; i++) {
      NonTXEntry entry = (NonTXEntry) recoveredRegion.getEntry("key" + i);
      tagsFromCrf[i] = entry.getRegionEntry().getVersionStamp().asVersionTag();
      LogWriterUtils.getLogWriter()
          .info("crfTag[" + i + "]=" + tagsFromCrf[i] + ",value=" + entry.getValue());
    }

    // Make sure the version tags from the krf and the crf match.
    for (int i = 0; i < 3; i++) {
      assertThat(tagsFromKrf[i]).isEqualTo(tagsFromCrf[i]);
    }
  }
Test that when we generate a krf, we write the version tag that matches the entry in the crf.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/persistence/PersistentRVVRecoveryDUnitTest.java