/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache;

import static java.lang.System.out;
import static org.apache.geode.internal.cache.DistributedCacheOperation.SLOW_DISTRIBUTION_MS;
import static org.apache.geode.internal.cache.InitialImageOperation.GIITestHookType.DuringPackingImage;
import static org.apache.geode.internal.cache.InitialImageOperation.getGIITestHookForCheckingPurpose;
import static org.apache.geode.internal.cache.InitialImageOperation.resetGIITestHook;
import static org.apache.geode.internal.cache.InitialImageOperation.setGIITestHook;
import static org.apache.geode.test.dunit.Host.getHost;
import static org.apache.geode.test.dunit.LogWriterUtils.getLogWriter;
import static org.apache.geode.test.dunit.Wait.pause;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.junit.Test;

import org.apache.geode.DataSerializer;
import org.apache.geode.GemFireIOException;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.Scope;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.cache.DestroyOperation.DestroyMessage;
import org.apache.geode.internal.cache.DistributedTombstoneOperation.TombstoneMessage;
import org.apache.geode.internal.cache.InitialImageOperation.GIITestHook;
import org.apache.geode.internal.cache.InitialImageOperation.GIITestHookType;
import org.apache.geode.internal.cache.InitialImageOperation.RequestImageMessage;
import org.apache.geode.internal.cache.UpdateOperation.UpdateMessage;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.persistence.DiskStoreID;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.util.internal.GeodeGlossary;


public class GIIDeltaDUnitTest extends JUnit4CacheTestCase {

  VM P; // GII provider
  VM R; // GII requester
  InternalDistributedMember R_ID; // DistributedMember ID of R

  private static final long MAX_WAIT = 30000;
  // protected static String REGION_NAME = GIIDeltaDUnitTest.class.getSimpleName()+"_Region";
  protected static String REGION_NAME = "_Region";
  final String expectedExceptions = GemFireIOException.class.getName();
  protected IgnoredException expectedEx;
  static Object giiSyncObject = new Object();

  public GIIDeltaDUnitTest() {
    super();
  }

  @Override
  public final void postSetUp() throws Exception {
    Invoke.invokeInEveryVM(GIIDeltaDUnitTest.class, "setRegionName",
        new Object[] {getUniqueName()});
    setRegionName(getUniqueName());
  }

  public static void setRegionName(String testName) {
    REGION_NAME = testName + "_Region";
  }

  @Override
  public final void preTearDownCacheTestCase() throws Exception {
    P.invoke(GIIDeltaDUnitTest::resetSlowGII);
    R.invoke(GIIDeltaDUnitTest::resetSlowGII);
    P.invoke(InitialImageOperation::resetAllGIITestHooks);
    R.invoke(InitialImageOperation::resetAllGIITestHooks);
    changeUnfinishedOperationLimit(R, 10000);
    changeForceFullGII(R, false, false);
    changeForceFullGII(P, false, false);
    P = null;
    R = null;
  }

  @Override
  public final void postTearDownCacheTestCase() throws Exception {
    // clean up the test hook, which can be moved to CacheTestCase
    DistributedCacheOperation.SLOW_DISTRIBUTION_MS = 0;
    if (expectedEx != null) {
      expectedEx.remove();
    }
  }

  // private VersionTag getVersionTag(VM vm, final String key) {
  // SerializableCallable getVersionTag = new SerializableCallable("verify recovered entry") {
  // public Object call() {
  // VersionTag tag = CCRegion.getVersionTag(key);
  // return tag;
  //
  // }
  // };
  // return (VersionTag)vm.invoke(getVersionTag);
  // }

  public InternalDistributedMember getDistributedMemberID(VM vm) {
    SerializableCallable getID = new SerializableCallable("get member id") {
      @Override
      public Object call() {
        return getCache().getDistributedSystem().getDistributedMember();
      }
    };
    InternalDistributedMember id = (InternalDistributedMember) vm.invoke(getID);
    return id;
  }

  protected void prepareForEachTest() {
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    VM vm1 = host.getVM(1);

    createDistributedRegion(vm0);
    createDistributedRegion(vm1);
    assignVMsToPandR(vm0, vm1);
    // from now on, use P and R as
    // vmhttps://wiki.gemstone.com/display/gfepersistence/DeltaGII+Spec+for+8.0
    expectedEx = IgnoredException.addIgnoredException(expectedExceptions);
  }

  // these steps are shared by all test cases
  private void prepareCommonTestData(int p_version) {
    // operation P1: region.put("key") at P. After the operation, region version for P becomes 1
    R_ID = getDistributedMemberID(R);

    assertDeltaGIICountBeZero(P);
    assertDeltaGIICountBeZero(R);
    doOnePut(P, 1, "key1");
    doOnePut(R, 1, "key2");
    doOnePut(R, 2, "key5");

    createConflictOperationsP2R3();

    for (int i = 3; i <= p_version; i++) {
      switch (i) {
        case 3:
          doOneDestroy(P, 3, "key1");
          break;
        case 4:
          doOneDestroy(P, 4, "key2");
          break;
        case 5:
          doOnePut(P, 5, "key1");
          break;
        case 6:
          doOnePut(P, 6, "key3");
        default:
          // donothing
      }
    }
    // at this moment, cache has key1, key3, key5
  }

  private void createConflictOperationsP2R3() {
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] blocklist = {2, 3};

    P.invoke(() -> GIIDeltaDUnitTest.slowGII(blocklist));
    R.invoke(() -> GIIDeltaDUnitTest.slowGII(blocklist));
    AsyncInvocation<Void> async1 = doOnePutAsync(P, 2, "key1");
    AsyncInvocation<Void> async2 = doOnePutAsync(R, 3, "key1");

    // wait for the local puts are done at P & R before distribution
    waitForToVerifyRVV(P, memberP, 2, null, 0); // P's rvv=p2, gc=0
    waitForToVerifyRVV(P, memberR, 2, null, 0); // P's rvv=r2, gc=0
    waitForToVerifyRVV(R, memberP, 1, null, 0); // P's rvv=p2, gc=0
    waitForToVerifyRVV(R, memberR, 3, null, 0); // P's rvv=r2, gc=0

    // new Object[] { memberP, 2, 3, 0, 0, false }
    P.invoke(GIIDeltaDUnitTest::resetSlowGII);
    R.invoke(GIIDeltaDUnitTest::resetSlowGII);

    // should wait for async calls to finish before going on
    checkAsyncCall(async1);
    checkAsyncCall(async2);

    // verify RVVs
    waitForToVerifyRVV(P, memberP, 2, null, 0); // P's rvv=p2, gc=0
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0
    waitForToVerifyRVV(R, memberP, 2, null, 0); // P's rvv=p2, gc=0
    waitForToVerifyRVV(R, memberR, 3, null, 0); // P's rvv=r3, gc=0

    // verify P won the conflict check
    waitToVerifyKey(P, "key1", generateValue(P));
    waitToVerifyKey(R, "key1", generateValue(P));
  }

  private void createUnfinishedOperationsR4R5() {
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    // let r6 to succeed, r4,r5 to be blocked
    R.invoke(() -> GIIDeltaDUnitTest.slowGII(exceptionlist));
    AsyncInvocation<Void> async1 = doOnePutAsync(R, 4, "key4");
    waitForToVerifyRVV(R, memberR, 4, null, 0); // P's rvv=r4, gc=0

    AsyncInvocation<Void> async2 = doOneDestroyAsync(R, 5, "key5");
    waitForToVerifyRVV(R, memberR, 5, null, 0); // P's rvv=r5, gc=0

    doOnePut(R, 6, "key1"); // r6 will pass
    waitForToVerifyRVV(R, memberR, 6, null, 0); // P's rvv=r6, gc=0

    // P should have exception list R4,R5 now
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6(3-6), gc=0
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. Each does a few operations to make RVV=P7,R6,
   * RVVGC=P4,R0 for both members. vm1 becomes offline then restarts. The deltaGII should only
   * exchange RVV. No need to send data from vm0 to vm1.
   */
  @Test
  public void testDeltaGIIWithSameRVV() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(6);

    // force tombstone GC to let RVVGC to become P4:R0
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0

    // let r4,r5,r6 to succeed
    doOnePut(R, 4, "key4");
    doOneDestroy(R, 5, "key5");
    doOnePut(R, 6, "key1");

    // let p7 to succeed
    doOnePut(P, 7, "key1");

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

    waitForToVerifyRVV(R, memberP, 7, null, 4); // P's rvv=p7, gc=4
    waitForToVerifyRVV(R, memberR, 6, null, 0); // P's rvv=r6, gc=0
    // now the rvv and rvvgc at P and R should be the same

    // save R's rvv in byte array, check if it will be fullGII
    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    // shutdown R and restart
    closeCache(R);

    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);
    createDistributedRegion(R);
    waitForToVerifyRVV(R, memberP, 7, null, 4); // P's rvv=p7, gc=4
    waitForToVerifyRVV(R, memberR, 6, null, 0); // P's rvv=r6, gc=0

    RegionVersionVector p_rvv = getRVV(P);
    RegionVersionVector r_rvv = getRVV(R);
    assertSameRVV(p_rvv, r_rvv); // after gii, rvv should be the same

    // If fullGII, the key size in gii chunk is 4, i.e. key1,key3,key4,key5(tombstone). key2 is
    // GCed.
    // If delta GII, the key size should be 0
    verifyDeltaSizeFromStats(R, 0, 1);
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. create some exception list. Before GII, P's RVV is
   * P7,R6(3-6), R's RVV is P6,R6, RVVGC are both P4,R0 vm1 becomes offline then restarts.
   * https://wiki.gemstone.com/display/gfepersistence/DeltaGII+Spec+for+8.0 The deltaGII should send
   * delta to R, revoke unfinished opeation R4,R5
   */
  @Test
  public void testDeltaGIIWithExceptionList() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(6);
    VersionTag expect_tag = getVersionTag(R, "key5");

    // force tombstone GC to let RVVGC to become P4:R0
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0

    createUnfinishedOperationsR4R5();

    // now P's cache still only has key1, key3, key5
    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);
    // p7 only apply at P
    doOnePut(P, 7, "key1");

    // restart and gii
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);
    createDistributedRegion(R);
    waitForToVerifyRVV(P, memberP, 7, null, 4); // P's rvv=p7, gc=4
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6, 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

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

    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    // If delta GII, the key size should be 2, i.e. P7(key1) and (key5(T) which is unfinished
    // operation)
    verifyDeltaSizeFromStats(R, 2, 1);
    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    VersionTag tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));

    // restart P, since R has received exceptionlist R4,R5 from P
    closeCache(P);
    createDistributedRegion(P);
    waitForToVerifyRVV(P, memberP, 7, null, 4); // P's rvv=p7, gc=4
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6, gc=0
    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key4 is removed as
    // unfinished op
    // If deltaGII, the key size should be 0
    verifyDeltaSizeFromStats(P, 0, 1);
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. create some exception list. Before GII, P's RVV is
   * P6,R6(3-6), R's RVV is P6,R6, RVVGC are both P4,R0 vm1 becomes offline then restarts. The
   * deltaGII should send delta which only contains unfinished operation R4,R5
   */
  @Test
  public void testDeltaGIIWithOnlyUnfinishedOp() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(6);
    VersionTag expect_tag = getVersionTag(R, "key5");

    // force tombstone GC to let RVVGC to become P4:R0
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0

    createUnfinishedOperationsR4R5();

    // now P's cache still only has key1, key3, key5
    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);

    // restart and gii
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);
    createDistributedRegion(R);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6, gc=0
    waitForToVerifyRVV(R, memberP, 6, null, 4); // R's rvv=p6, gc=4
    waitForToVerifyRVV(R, memberR, 6, exceptionlist, 0); // R's rvv=r6, gc=0

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

    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    // If delta GII, the key size should be 1 (key5(T) which is unfinished operation)
    verifyDeltaSizeFromStats(R, 1, 1);
    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    VersionTag tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));

    // restart P, since R has received exceptionlist R4,R5 from P
    closeCache(P);
    createDistributedRegion(P);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6, gc=0
    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key4 is removed as
    // unfinished op
    // If deltaGII, the key size should be 0
    verifyDeltaSizeFromStats(P, 0, 1);

    // restart R, to make sure the unfinished op is handled correctly
    forceGC(R, 1); // for bug 47616
    waitForToVerifyRVV(P, memberR, 6, null, 5); // P's rvv=R6, gc=5
    waitForToVerifyRVV(R, memberR, 6, null, 5); // P's rvv=R6, gc=5

    closeCache(R);
    createDistributedRegion(R);
    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key4 is removed as
    // unfinished op
    // If deltaGII, the key size should be 0
    verifyDeltaSizeFromStats(R, 0, 1);
    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));
  }

  /**
   * This is to test a race condition for bug#47616 vm0 and vm1 are peers, each holds a DR. create
   * some exception list. Before GII, P's RVV is P6,R6(3-6), R's RVV is P6,R6, RVVGC are both P4,R0
   * vm1 becomes offline then restarts. The deltaGII should send delta which only contains
   * unfinished opeation R4,R5
   */
  @Test
  public void testDeltaGIIWithOnlyUnfinishedOp_GCAtR() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(6);
    VersionTag expect_tag = getVersionTag(R, "key5");

    // force tombstone GC to let RVVGC to become P4:R0
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0

    createUnfinishedOperationsR4R5();
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6, gc=0

    // 2
    R.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        Mycallback myAfterRequestRVV = new Mycallback(GIITestHookType.AfterRequestRVV, REGION_NAME);
        InitialImageOperation.setGIITestHook(myAfterRequestRVV);
      }
    });

    // now P's cache still only has key1, key3, key5
    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);

    // restart and gii
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);
    AsyncInvocation<Void> async3 = createDistributedRegionAsync(R);

    // 2
    waitForCallbackStarted(R, GIITestHookType.AfterRequestRVV);
    forceGC(R, 1);
    R.invoke(() -> InitialImageOperation.resetGIITestHook(GIITestHookType.AfterRequestRVV, true));

    async3.join(MAX_WAIT);

    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    VersionTag tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));
    verifyTombstoneExist(R, "key5", false, false);

    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    // If delta GII, the key size should be 1 (key5(T) which is unfinished operation)
    verifyDeltaSizeFromStats(R, 1, 1);
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. create some exception list. Then shutdown R. Do
   * tombstone GC at P only. Before GII, P's RVV is P7,R6(3-6), RVVGC is P4,R0; R's RVV is P6,R6,
   * RVVGC are both P0,R0 vm1 becomes offline then restarts.
   * https://wiki.gemstone.com/display/gfepersistence/DeltaGII+Spec+for+8.0 The deltaGII should send
   * delta to R, revoke unfinished opeation R4,R5
   */
  @Test
  public void testDeltaGIIWithDifferentRVVGC() throws Throwable {
    final String testcase = "testDeltaGIIWithExceptionList";
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

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

    createUnfinishedOperationsR4R5();

    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);

    // force tombstone GC to let RVVGC to become P4:R0
    changeTombstoneTimout(R, MAX_WAIT);
    changeTombstoneTimout(P, MAX_WAIT);
    Wait.pause((int) MAX_WAIT);
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 6, null, 4); // P's rvv=p6, gc=4

    // p7 only apply at P
    doOnePut(P, 7, "key1");
    waitForToVerifyRVV(P, memberP, 7, null, 4); // P's rvv=p7, gc=4
    verifyTombstoneExist(P, "key2", false, false);

    // restart R and gii
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);
    createDistributedRegion(R);
    waitForToVerifyRVV(R, memberP, 7, null, 4); // R's rvv=p7, gc=4
    waitForToVerifyRVV(R, memberR, 6, exceptionlist, 0); // R's rvv=r6, gc=0

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

    // If fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    // If delta GII, the key size should be 2, i.e. P7 (key1) and (key5(T) which is unfinished
    // operation)
    verifyDeltaSizeFromStats(R, 2, 1);
    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    VersionTag tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));
    verifyTombstoneExist(R, "key2", false, false);
    verifyTombstoneExist(P, "key2", false, false);
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. Let provider to have higher RVVGC than requester's RVV
   * It should trigger fullGII
   *
   * This test also verify tombstoneGC happened in middle of GII, but BEFORE GII thread got GIILock.
   * i.e. before GII, P's RVVGC=P0,R0, upon received RequestImageMessage, it becomes P4,R0 it should
   * cause the fullGII.
   */
  @Test
  public void testFullGIITriggeredByHigherRVVGC() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(3);
    waitForToVerifyRVV(P, memberP, 3, null, 0); // P's rvv=p3, gc=0

    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
    // 7
    P.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        Mycallback myAfterReceivedRequestImage =
            new Mycallback(GIITestHookType.AfterReceivedRequestImage, REGION_NAME);
        InitialImageOperation.setGIITestHook(myAfterReceivedRequestImage);
      }
    });

    // restart R and gii, it will be blocked at test hook
    AsyncInvocation<Void> async3 = createDistributedRegionAsync(R);
    // 7
    waitForCallbackStarted(P, GIITestHookType.AfterReceivedRequestImage);

    // force tombstone GC to let RVVGC to become P4:R0, but R already sent its old RVV/RVVGC over
    // this tombstone GC happens BEFORE GII thread got the GIILock, so it will be fullGCC
    forceGC(P, 2);
    waitForToVerifyRVV(P, memberP, 7, null, 4); // P's rvv=p7, gc=4
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6(3-6), gc=0
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, true);

    // let GII continue
    P.invoke(() -> InitialImageOperation.resetGIITestHook(GIITestHookType.AfterReceivedRequestImage,
        true));
    async3.join(MAX_WAIT);

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

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

    // In fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    verifyDeltaSizeFromStats(R, 3, 0);
  }

  /**
   * vm0(P), vm1(R), vm2(T) are peers, each holds a DR. shutdown vm1(R), vm2(T) Let provider to have
   * higher RVVGC than requester's RVV when vm1 and vm2 are offline Restart R, It should trigger
   * fullGII If R did not save the rvvgc, restarted R will have a way smaller rvvgc (maybe the same
   * as T's) Let T requests GII from R, it wll become deltaGII, which is wrong.
   */
  @Test
  public void testSavingRVVGC() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

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

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(3);
    waitForToVerifyRVV(P, memberP, 3, null, 0); // P's rvv=p3, gc=0

    waitForToVerifyRVV(R, memberP, 3, null, 0); // R's rvv=p3, gc=0
    waitForToVerifyRVV(T, memberP, 3, null, 0); // T's rvv=p3, gc=0

    closeCache(T);
    closeCache(R);

    // p4-7 only apply at P
    doOneDestroy(P, 4, "key2");
    doOnePut(P, 5, "key1");
    doOnePut(P, 6, "key3");
    doOnePut(P, 7, "key1");
    // force tombstone GC to let RVVGC to become P4:R0
    forceGC(P, 2);

    // restart R and gii, it will be blocked at test hook
    createDistributedRegion(R);

    waitForToVerifyRVV(P, memberP, 7, null, 4); // P's rvv=p7, gc=4
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0
    waitForToVerifyRVV(R, memberP, 7, null, 4); // R's rvv=p7, gc=4
    waitForToVerifyRVV(R, memberR, 3, null, 0); // R's rvv=r3, gc=0

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

    // In fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    verifyDeltaSizeFromStats(R, 3, 0);
    verifyTombstoneExist(P, "key2", false, false);
    verifyTombstoneExist(R, "key2", false, false);

    closeCache(P);
    closeCache(R);
    createDistributedRegion(R);
    waitForToVerifyRVV(R, memberP, 7, null, 4); // R's rvv=p7, gc=4
    waitForToVerifyRVV(R, memberR, 3, null, 0); // R's rvv=r3, gc=0
    verifyTombstoneExist(R, "key2", false, false);

    createDistributedRegion(T);
    waitForToVerifyRVV(T, memberP, 7, null, 4); // R's rvv=p7, gc=4
    waitForToVerifyRVV(T, memberR, 3, null, 0); // R's rvv=r3, gc=0
    // In fullGII, the key size in gii chunk is 3, i.e. key1,key3,key5. key2 is GCed.
    verifyDeltaSizeFromStats(T, 3, 0);
    verifyTombstoneExist(T, "key2", false, false);
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. create some exception list. Not to do any tombstone GC
   * Before GII, P's RVV is P7,R6(3-6), R's RVV is P6,R6, RVVGC are both P0,R0 vm1 becomes offline
   * then restarts. https://wiki.gemstone.com/display/gfepersistence/DeltaGII+Spec+for+8.0 The
   * deltaGII should send delta to R, revoke unfinished opeation R4,R5
   */
  @Test
  public void testDeltaGIIWithoutRVVGC() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(6);
    waitForToVerifyRVV(P, memberP, 6, null, 0); // P's rvv=p6, gc=0
    waitForToVerifyRVV(P, memberR, 3, null, 0); // P's rvv=r3, gc=0
    VersionTag expect_tag = getVersionTag(R, "key5");

    createUnfinishedOperationsR4R5();

    // now P's cache still only has key1, key3
    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);
    // p7 only apply at P
    doOnePut(P, 7, "key1");

    // restart R and gii
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, false);
    createDistributedRegion(R);

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

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

    // If fullGII, the key size in gii chunk is 4.
    // tombstone counts in deltaSize, so without rvvgc, it's 4: key1, key2(tombstone), key3, key5
    // In delta GII, it should be 1, i.e. P7 (key1) and (key5(T) which is unfinished operation)
    verifyDeltaSizeFromStats(R, 2, 1);
    // verify unfinished op for key5 is revoked
    waitToVerifyKey(R, "key5", generateValue(R));
    VersionTag tag = getVersionTag(R, "key5");
    assertTrue(expect_tag.equals(tag));
  }

  /**
   * vm0 and vm1 are peers, each holds a DR. unifinished P8: destroy(key1), finished P9: put(key3).
   * Shutdown R, then GC tombstones at P. P's RVV=P9,R6(3-6), RVVGC=P8,R0, R's RVV=P9(7-9),R6, RVV
   * It should trigger fullGII
   */
  @Test
  public void testFullGIINotDorminatedByProviderRVVGC() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);
    final long[] exceptionlist = {4, 5};

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(3);

    createUnfinishedOperationsR4R5();

    waitForToVerifyRVV(P, memberP, 3, null, 0); // P's rvv=p3, gc=0
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6(3-6), gc=0
    waitForToVerifyRVV(R, memberP, 3, null, 0); // R's rvv=p3, gc=0
    waitForToVerifyRVV(R, memberR, 6, null, 0); // R's rvv=r6, gc=0

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

    final long[] exceptionlist2 = {8};
    // let p9 to succeed, p8 to be blocked
    P.invoke(() -> GIIDeltaDUnitTest.slowGII(exceptionlist2));
    AsyncInvocation<Void> async1 = doOneDestroyAsync(P, 8, "key1");
    waitForToVerifyRVV(P, memberP, 8, null, 0);
    doOnePut(P, 9, "key3");
    waitForToVerifyRVV(P, memberP, 9, null, 0);
    waitForToVerifyRVV(R, memberP, 9, exceptionlist2, 0);

    byte[] R_rvv_bytes = getRVVByteArray(R, REGION_NAME);
    closeCache(R);
    forceGC(P, 3);
    // now P's RVV=P9,R6(3-6), RVVGC=P8,R0, R's RVV=P9(7-9), R6
    waitForToVerifyRVV(P, memberP, 9, null, 8); // P's rvv=p9, gc=8
    waitForToVerifyRVV(P, memberR, 6, exceptionlist, 0); // P's rvv=r6(3-6), gc=0
    P.invoke(GIIDeltaDUnitTest::resetSlowGII);

    // restart and gii, R's rvv should be the same as P's
    checkIfFullGII(P, REGION_NAME, R_rvv_bytes, true);
    createDistributedRegion(R);
    waitForToVerifyRVV(R, memberP, 9, null, 8); // R's rvv=p9, gc=8
    waitForToVerifyRVV(R, memberR, 6, exceptionlist, 0); // R's rvv=r6, gc=0

    RegionVersionVector p_rvv = getRVV(P);
    RegionVersionVector r_rvv = getRVV(R);
    assertSameRVV(p_rvv, r_rvv); // after gii, rvv should be the same

    // In fullGII, the key size in gii chunk is 2. They are: key3, key5
    verifyDeltaSizeFromStats(R, 2, 0);
  }

  /**
   * Let R4, R5 unfinish, but R5 is the last operation from R. So P's RVV is still P:x,R3, without
   * exception list. But actually R4, R5 are unfinished ops by all means.
   */
  @Test
  public void testUnfinishedOpsWithoutExceptionList() throws Throwable {
    prepareForEachTest();
    final DiskStoreID memberP = getMemberID(P);
    final DiskStoreID memberR = getMemberID(R);

    assertEquals(0, DistributedCacheOperation.SLOW_DISTRIBUTION_MS);
    prepareCommonTestData(6);
    VersionTag expect_tag = getVersionTag(R, "key5");

    final long[] exceptionlist = {4, 5};
