@Test
  public void testTombstoneGCInMiddleOfGII() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    Host host = getHost(0);
    VM T = host.getVM(2);
    createDistributedRegion(T);
    final DiskStoreID memberT = getMemberID(T);
    closeCache(T);

    assertEquals(0, SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(3);
    waitForToVerifyRVV(P, memberP, 3, null, 0); // P's rvv=p3, gc=0
    VersionTag expect_tag = getVersionTag(R, "key5");

    createUnfinishedOperationsR4R5();
    waitForToVerifyRVV(R, memberP, 3, null, 0); // R's rvv=p3, gc=0
    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);

    // p4-7 only apply at P
    doOneDestroy(P, 4, "key2");
    doOnePut(P, 5, "key1");
    doOnePut(P, 6, "key3");
    doOnePut(P, 7, "key1");

    // add test hook
    // 8
    P.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        Mycallback myDuringPackingImage =
            new Mycallback(DuringPackingImage, REGION_NAME);
        setGIITestHook(myDuringPackingImage);
      }
    });

    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);

    // restart R and gii, it will be blocked at test hook
    AsyncInvocation<Void> async3 = createDistributedRegionAsync(R);
    // restart R and gii, it will be blocked at test hook
    AsyncInvocation<Void> async4 = createDistributedRegionAsync(T);
    // 8
    waitForCallbackStarted(P, DuringPackingImage);
    WaitCriterion ev = new WaitCriterion() {
      @Override
      public boolean done() {
        int count = getDeltaGIICount(P);
        return (count == 2);
      }

      @Override
      public String description() {
        return null;
      }
    };

    GeodeAwaitility.await().untilAsserted(ev);
    int count = getDeltaGIICount(P);
    assertEquals(2, count);

    // force tombstone GC to let RVVGC to become P4:R0, but R already sent its old RVV/RVVGC over
    // this tombstone GC happens AFTER GII thread got the GIILock, so it will be ignored since GII
    // is ongoing
    changeTombstoneTimout(R, MAX_WAIT);
    changeTombstoneTimout(P, MAX_WAIT);
    changeTombstoneTimout(T, MAX_WAIT);
    pause((int) MAX_WAIT);
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 7, null, 0); // P's rvv=p7, gc=0

    // let GII continue
    P.invoke(
        () -> resetGIITestHook(DuringPackingImage, false));
    P.invoke(
        () -> resetGIITestHook(DuringPackingImage, true));

    WaitCriterion ev2 = new WaitCriterion() {
      @Override
      public boolean done() {
        int count = getDeltaGIICount(P);
        return (count == 0);
      }

      @Override
      public String description() {
        return null;
      }
    };
    GeodeAwaitility.await().untilAsserted(ev2);
    count = getDeltaGIICount(P);
    assertEquals(0, count);
    verifyTombstoneExist(P, "key2", true, true); // expect key2 is still tombstone during and after
                                                 // GIIs
    verifyTombstoneExist(R, "key2", true, true); // expect key2 is still tombstone during and after
                                                 // GIIs
    verifyTombstoneExist(T, "key2", true, true); // expect key2 is still tombstone during and after
                                                 // GIIs

    forceGC(P, 1); // trigger to GC the tombstones in expired queue
    async3.join(MAX_WAIT);
    async4.join(MAX_WAIT);

    // after GII, tombstone GC happened
    waitForToVerifyRVV(P, memberP, 7, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6(3-6), gc=0

    waitForToVerifyRVV(R, memberP, 7, null, 4); // R's rvv=p7, gc=4
    waitForToVerifyRVV(R, memberR, 6, exceptionlist, 0); // R's rvv=r6, gc=0

    verifyTombstoneExist(P, "key2", false, true); // expect tombstone key2 is GCed at P
    verifyTombstoneExist(R, "key2", false, true); // expect tombstone key2 is GCed at R
    verifyTombstoneExist(T, "key2", false, true); // T got everything from P

    // do a put from T
    doOnePut(T, 1, "key1");

    RegionVersionVector p_rvv = getRVV(P);
    RegionVersionVector r_rvv = getRVV(R);
    RegionVersionVector t_rvv = getRVV(R);
    assertSameRVV(p_rvv, r_rvv);
    assertSameRVV(t_rvv, r_rvv);

    // If fullGII, the key size in gii chunk is 4, i.e. key1,key3,key5,key2 is a tombstone.
    // If delta GII, it should be 4, (key1, key2, key3) and (key5(T) which is unfinished operation)
    verifyDeltaSizeFromStats(R, 4, 1);
    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    VersionTag tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));
    // P.invoke(() -> InitialImageOperation.resetAllGIITestHooks());
  }
vm0 and vm1 are peers, each holds a DR. create some exception list. Then shutdown R. Before
GII, P's RVV is P7,R6(3-6), RVVGC is P0,R0; R's RVV is P3,R6, RVVGC is P0,R0 vm1 becomes
offline then restarts. Use testHook to pause the GII, then do tombstone GC at P only. The
deltaGII should send correct tombstonedelta to R, revoke unfinished opeation R4,R5

There's member T doing GII from P at the same time.

In this test, GII thread will get the GIILock before tombstone GC, so tombstone GC should wait
for all GIIs to finish
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/GIIDeltaDUnitTest.java