/*
 * 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.ha;

import static java.lang.Thread.yield;
import static org.apache.geode.internal.cache.ha.HARegionQueue.NON_BLOCKING_HA_QUEUE;
import static org.apache.geode.internal.cache.ha.HARegionQueue.getHARegionQueueInstance;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Assert.assertEquals;
import static org.apache.geode.test.dunit.Assert.assertNotNull;
import static org.apache.geode.test.dunit.Assert.assertNull;
import static org.apache.geode.test.dunit.Assert.assertTrue;
import static org.apache.geode.test.dunit.Assert.fail;
import static org.apache.geode.test.dunit.ThreadUtils.join;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.mockito.AdditionalAnswers;

import org.apache.geode.LogWriter;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.HARegion;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

@Category({ClientSubscriptionTest.class})
public class HARegionQueueDUnitTest extends JUnit4DistributedTestCase {

  private static volatile boolean toCnt = true;
  private static volatile Thread createQueuesThread;

  private static InternalCache cache = null;
  private static HARegionQueue hrq = null;
  private static Thread[] opThreads;

  private VM vm0 = null;
  private VM vm1 = null;
  private VM vm3 = null;
  private VM vm2 = null;

  /**
   * get the VM's
   */
  @Override
  public final void postSetUp() throws Exception {
    final Host host = Host.getHost(0);
    vm0 = host.getVM(0);
    vm1 = host.getVM(1);
    vm2 = host.getVM(2);
    vm3 = host.getVM(3);
    vm0.invoke(() -> HARegionQueueDUnitTest.toCnt = true);
    vm1.invoke(() -> HARegionQueueDUnitTest.toCnt = true);
    vm2.invoke(() -> HARegionQueueDUnitTest.toCnt = true);
    vm3.invoke(() -> HARegionQueueDUnitTest.toCnt = true);
  }

  /**
   * close the cache in tearDown
   */
  @Override
  public final void preTearDown() throws Exception {
    vm0.invoke(() -> HARegionQueueDUnitTest.closeCache());
    vm1.invoke(() -> HARegionQueueDUnitTest.closeCache());
    vm2.invoke(() -> HARegionQueueDUnitTest.closeCache());
    vm3.invoke(() -> HARegionQueueDUnitTest.closeCache());

    cache = null;
    hrq = null;
    opThreads = null;
  }

  /**
   * create cache
   */
  private InternalCache createCache() throws CacheException {
    Properties props = new Properties();
    DistributedSystem ds = getSystem(props);
    ds.disconnect();
    ds = getSystem(props);
    InternalCache cache = null;
    cache = (InternalCache) CacheFactory.create(ds);
    if (cache == null) {
      // TODO: never throw an anonymous inner class
      throw new CacheException("CacheFactory.create() returned null ") {};
    }
    return cache;
  }

  /**
   * 1) Create mirrored HARegion region1 in VM1 and VM2 2) do a put in VM1 3) assert that the put
   * has not propagated from VM1 to VM2 4) do a put in VM2 5) assert that the value in VM1 has not
   * changed to due to put in VM2 6) assert put in VM2 was successful by doing a get
   */
  @Test
  public void testLocalPut() throws Exception {
    vm0.invoke(() -> HARegionQueueDUnitTest.createRegion());
    vm1.invoke(() -> HARegionQueueDUnitTest.createRegion());
    vm0.invoke(() -> HARegionQueueDUnitTest.putValue1());
    vm1.invoke(() -> HARegionQueueDUnitTest.getNull());
    vm1.invoke(() -> HARegionQueueDUnitTest.putValue2());
    vm0.invoke(() -> HARegionQueueDUnitTest.getValue1());
    vm1.invoke(() -> HARegionQueueDUnitTest.getValue2());
  }

  /**
   * 1) Create mirrored HARegion region1 in VM1 and VM2 2) do a put in VM1 3) assert that the put
   * has not propagated from VM1 to VM2 4) do a put in VM2 5) assert that the value in VM1 has not
   * changed to due to put in VM2 6) assert respective puts the VMs were successful by doing a get
   * 7) localDestroy key in VM1 8) assert key has been destroyed in VM1 9) assert key has not been
   * destroyed in VM2
   */
  @Test
  public void testLocalDestroy() throws Exception {
    vm0.invoke(() -> HARegionQueueDUnitTest.createRegion());
    vm1.invoke(() -> HARegionQueueDUnitTest.createRegion());
    vm0.invoke(() -> HARegionQueueDUnitTest.putValue1());
    vm1.invoke(() -> HARegionQueueDUnitTest.getNull());
    vm1.invoke(() -> HARegionQueueDUnitTest.putValue2());
    vm0.invoke(() -> HARegionQueueDUnitTest.getValue1());
    vm1.invoke(() -> HARegionQueueDUnitTest.getValue2());
    vm0.invoke(() -> HARegionQueueDUnitTest.destroy());
    vm0.invoke(() -> HARegionQueueDUnitTest.getNull());
    vm1.invoke(() -> HARegionQueueDUnitTest.getValue2());
  }

  /**
   * 1) Create mirrored HARegion region1 in VM1 2) do a put in VM1 3) get the value in VM1 to assert
   * put has happened successfully 4) Create mirrored HARegion region1 in VM2 5) do a get in VM2 to
   * verify that value was got through GII 6) do a put in VM2 7) assert put in VM2 was successful
   */
  @Test
  public void testGII() throws Exception {
    vm0.invoke(() -> HARegionQueueDUnitTest.createRegion());
    vm0.invoke(() -> HARegionQueueDUnitTest.putValue1());
    vm0.invoke(() -> HARegionQueueDUnitTest.getValue1());
    vm1.invoke(() -> HARegionQueueDUnitTest.createRegion());
    vm1.invoke(() -> HARegionQueueDUnitTest.getValue1());
    vm1.invoke(() -> HARegionQueueDUnitTest.putValue2());
    vm1.invoke(() -> HARegionQueueDUnitTest.getValue2());
  }

  /**
   * 1) Create mirrored HARegion region1 in VM1 2) do a put in VM1 3) get the value in VM1 to assert
   * put has happened successfully 4) Create mirrored HARegion region1 in VM2 5) do a get in VM2 to
   * verify that value was got through GII 6) do a put in VM2 7) assert put in VM2 was successful
   */
  @Test
  public void testQRM() throws Exception {
    vm0.invoke(() -> HARegionQueueDUnitTest.createRegionQueue());
    vm1.invoke(() -> HARegionQueueDUnitTest.createRegionQueue());
    vm0.invoke(() -> HARegionQueueDUnitTest.verifyAddingDispatchMesgs());

    vm1.invoke(() -> HARegionQueueDUnitTest.verifyDispatchedMessagesRemoved());
  }

  /**
   * Behaviour of take() has been changed for reliable messaging feature. Region queue take()
   * operation will no longer add to the Dispatch Message Map. Hence disabling the test - SUYOG
   *
   * Test for #35988 HARegionQueue.take() is not functioning as expected
   */
  @Ignore("TODO: this test was disabled")
  @Test
  public void testBugNo35988() throws Exception {
    CacheSerializableRunnable createQueue =
        new CacheSerializableRunnable("CreateCache, HARegionQueue and start thread") {
          @Override
          public void run2() throws CacheException {
            HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
            // TODO:ASIF: Bcoz of the QRM thread cannot take frequency below
            // 1 second , thus we need to carfully evaluate what to do. Though
            // in this case 1 second instead of 500 ms will work
            // System.getProperties().put("QueueRemovalThreadWaitTime", new Long(500));
            cache = test.createCache();
            cache.setMessageSyncInterval(1);
            HARegionQueueAttributes hrqa = new HARegionQueueAttributes();
            hrqa.setExpiryTime(300);
            try {
              hrq = HARegionQueue.getHARegionQueueInstance("testregion1", cache, hrqa,
                  HARegionQueue.NON_BLOCKING_HA_QUEUE, false);
              // Do 1000 putand 100 take in a separate thread
              hrq.put(new ConflatableObject(new Long(1), new Long(1),
                  new EventID(new byte[] {0}, 1, 1), false, "dummy"));
            } catch (Exception e) {
              throw new AssertionError(e);
            }
          }
        };
    vm0.invoke(createQueue);
    vm1.invoke(createQueue);

    vm0.invoke(new CacheSerializableRunnable("takeFromVm0") {
      @Override
      public void run2() throws CacheException {
        try {
          Conflatable obj = (Conflatable) hrq.take();
          assertNotNull(obj);
        } catch (Exception e) {
          throw new AssertionError(e);
        }
      }
    });

    vm1.invoke(new CacheSerializableRunnable("checkInVm1") {
      @Override
      public void run2() throws CacheException {
        WaitCriterion ev = new WaitCriterion() {
          @Override
          public boolean done() {
            yield(); // TODO is this necessary?
            return hrq.size() == 0;
          }

          @Override
          public String description() {
            return null;
          }
        };
        GeodeAwaitility.await().untilAsserted(ev);
      }
    });

  }

  /**
   * create a client with 2 regions sharing a common writer
   */
  private static void createRegion() throws Exception {
    HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
    cache = test.createCache();
    AttributesFactory factory = new AttributesFactory();
    factory.setScope(Scope.DISTRIBUTED_ACK);
    factory.setDataPolicy(DataPolicy.REPLICATE);

    // Mock the HARegionQueue and answer the input CachedDeserializable when updateHAEventWrapper is
    // called
    HARegionQueue harq = mock(HARegionQueue.class);
    when(harq.updateHAEventWrapper(any(), any(), any()))
        .thenAnswer(AdditionalAnswers.returnsSecondArg());

    HARegion.getInstance("HARegionQueueDUnitTest_region", (GemFireCacheImpl) cache, harq,
        factory.create());
  }

  private static void createRegionQueue() throws Exception {
    HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
    cache = test.createCache();
    /*
     * AttributesFactory factory = new AttributesFactory(); factory.setScope(Scope.DISTRIBUTED_ACK);
     * factory.setDataPolicy(DataPolicy.REPLICATE);
     */
    hrq = HARegionQueue.getHARegionQueueInstance("HARegionQueueDUnitTest_region", cache,
        HARegionQueue.NON_BLOCKING_HA_QUEUE, false);
    EventID id1 = new EventID(new byte[] {1}, 1, 1);
    EventID id2 = new EventID(new byte[] {1}, 1, 2);
    ConflatableObject c1 =
        new ConflatableObject("1", "1", id1, false, "HARegionQueueDUnitTest_region");
    ConflatableObject c2 =
        new ConflatableObject("2", "2", id2, false, "HARegionQueueDUnitTest_region");
    hrq.put(c1);
    hrq.put(c2);
  }

  private static void createRegionQueue2() throws Exception {
    HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
    cache = test.createCache();
    /*
     * AttributesFactory factory = new AttributesFactory(); factory.setScope(Scope.DISTRIBUTED_ACK);
     * factory.setDataPolicy(DataPolicy.REPLICATE);
     */
    HARegionQueueAttributes harqAttr = new HARegionQueueAttributes();
    harqAttr.setExpiryTime(3);
    hrq = HARegionQueue.getHARegionQueueInstance("HARegionQueueDUnitTest_region", cache, harqAttr,
        HARegionQueue.NON_BLOCKING_HA_QUEUE, false);
  }

  private static void clearRegion() {
    try {
      Iterator iterator = hrq.getRegion().keys().iterator();
      while (iterator.hasNext()) {
        hrq.getRegion().localDestroy(iterator.next());
      }
    } catch (Exception e) {
      fail("Exception occurred while trying to destroy region", e);
    }

  }

  private static void verifyAddingDispatchMesgs() {
    assertTrue(HARegionQueue.getDispatchedMessagesMapForTesting().isEmpty());
    hrq.addDispatchedMessage(new ThreadIdentifier(new byte[1], 1), 1);
    assertTrue(!HARegionQueue.getDispatchedMessagesMapForTesting().isEmpty());
  }

  private static void verifyDispatchedMessagesRemoved() {
    try {
      final Region region = hrq.getRegion();
      // wait until we have a dead server
      WaitCriterion ev = new WaitCriterion() {
        @Override
        public boolean done() {
          yield(); // TODO is this necessary?
          return region.get(new Long(0)) == null;
        }

        @Override
        public String description() {
          return null;
        }
      };
      GeodeAwaitility.await().untilAsserted(ev);

      /*
       * if (region.get(new Long(0)) != null) { fail("Expected message to have been deleted but it
       * is not deleted"); }
       */

      if (region.get(new Long(1)) == null) {
        fail("Expected message not to have been deleted but it is deleted");
      }

    } catch (Exception e) {
      fail("test failed due to an exception", e);
    }
  }

  /**
   * close the cache
   */
  private static void closeCache() {
    if (cache != null && !cache.isClosed()) {
      cache.close();
      cache.getDistributedSystem().disconnect();
    }
  }

  /**
   * do puts on key-1
   */
  private static void putValue1() {
    try {
      Region r1 = cache.getRegion("/HARegionQueueDUnitTest_region");
      r1.put("key-1", "value-1");
    } catch (Exception ex) {
      fail("failed while region.put()", ex);
    }
  }

  private static void putConflatables() {
    try {
      Region r1 = hrq.getRegion();
      for (int i = 1; i < 11; i++) {
        r1.put(new Long(i), new ConflatableObject("key" + i, "value" + i,
            new EventID(new byte[] {1}, 1, i), true, "HARegionQueueDUnitTest_region"));
      }
    } catch (Exception ex) {
      fail("failed while region.put()", ex);
    }
  }

  /**
   * verifies the data has been populated correctly after GII
   */
  private static void verifyMapsAndData() {
    try {
      HARegion r1 = (HARegion) hrq.getRegion();
      // region should not be null
      assertNotNull(" Did not expect the HARegion to be null but it is", r1);
      // it should have ten non null entries
      for (int i = 1; i < 11; i++) {
        assertNotNull(" Did not expect the entry to be null but it is", r1.get(new Long(i)));
      }
      // HARegionQueue should not be null
      assertNotNull(" Did not expect the HARegionQueue to be null but it is", hrq);

      Map conflationMap = hrq.getConflationMapForTesting();
      // conflationMap size should be greater than 0
      assertTrue(" Did not expect the conflationMap size to be 0 but it is",
          conflationMap.size() > 0);
      Map internalMap = (Map) conflationMap.get("HARegionQueueDUnitTest_region");
      // internal map should not be null. it should be present
      assertNotNull(" Did not expect the internalMap to be null but it is", internalMap);
      // get and verify the entries in the conflation map.
      for (int i = 1; i < 11; i++) {
        assertTrue(" Did not expect the entry not to be equal but it is",
            internalMap.get("key" + i).equals(new Long(i)));
      }
      Map eventMap = hrq.getEventsMapForTesting();
      // DACE should not be null
      assertNotNull(" Did not expect the result (DACE object) to be null but it is",
          eventMap.get(new ThreadIdentifier(new byte[] {1}, 1)));
      Set counterSet = hrq.getCurrentCounterSet(new EventID(new byte[] {1}, 1, 1));
      assertTrue(" excpected the counter set size to be 10 but it is not so",
          counterSet.size() == 10);
      long i = 1;
      Iterator iterator = counterSet.iterator();
      // verify the order of the iteration. it should be 1 - 10. The underlying
      // set is a LinkedHashSet
      while (iterator.hasNext()) {
        assertTrue(((Long) iterator.next()).longValue() == i);
        i++;
      }
      // The last dispactchde sequence Id should be -1 since no dispatch has
      // been made
      assertTrue(hrq.getLastDispatchedSequenceId(new EventID(new byte[] {1}, 1, 1)) == -1);

      // sleep for 8.0 seconds. Everythign should expire and everything should
      // be null and empty
      Thread.sleep(7500);

      for (int j = 1; j < 11; j++) {
        assertNull("expected the entry to be null since expiry time exceeded but it is not so",
            r1.get(new Long(j)));
      }

      internalMap = (Map) hrq.getConflationMapForTesting().get("HARegionQueueDUnitTest_region");

      assertNotNull(" Did not expect the internalMap to be null but it is", internalMap);
      assertTrue(
          "internalMap (conflation) should have been emptry since expiry of all entries has been exceeded but it is not so",
          internalMap.isEmpty());
      assertTrue(
          "eventMap should have been emptry since expiry of all entries has been exceeded but it is not so",
          eventMap.isEmpty());
      assertTrue(
          "counter set should have been emptry since expiry of all entries has been exceeded but it is not so",
          counterSet.isEmpty());

    } catch (Exception ex) {
      fail("failed while region.put()", ex);
    }
  }

  /**
   * do puts on key-1,value-2
   */
  private static void putValue2() {
    try {
      Region r1 = cache.getRegion("/HARegionQueueDUnitTest_region");
      r1.put("key-1", "value-2");
    } catch (Exception ex) {
      fail("failed while region.put()", ex);
    }
  }

  /**
   * do a get on region1
   */
  private static void getValue1() {
    try {
      Region r = cache.getRegion("/HARegionQueueDUnitTest_region");
      if (!(r.get("key-1").equals("value-1"))) {
        fail("expected value to be value-1 but it is not so");
      }

    } catch (Exception ex) {
      fail("failed while region.get()", ex);
    }
  }

  /**
   * do a get on region1
   */
  private static void getNull() {
    try {
      Region r = cache.getRegion("/HARegionQueueDUnitTest_region");
      if (!(r.get("key-1") == (null))) {
        fail("expected value to be null but it is not so");
      }

    } catch (Exception ex) {
      fail("failed while region.get()", ex);
    }
  }

  /**
   * do a get on region1
   */
  public static void getValue2() {
    try {
      Region r = cache.getRegion("/HARegionQueueDUnitTest_region");
      if (!(r.get("key-1").equals("value-2"))) {
        fail("expected value to be value-2 but it is not so");
      }

    } catch (Exception ex) {
      fail("failed while region.get()", ex);
    }
  }

  /**
   * destroy key-1
   */
  public static void destroy() {
    try {
      Region region1 = cache.getRegion("/HARegionQueueDUnitTest_region");
      region1.localDestroy("key-1");
    } catch (Exception e) {
      fail("test failed due to exception in destroy", e);
    }
  }

  /**
   * Tests the Non Blocking HARegionQueue by doing concurrent put /remove / take / peek , batch peek
   * operations in multiple regions. The test will have take/remove occuring in all the VMs. This
   * test is targetted to test for hang or exceptions in non blocking queue.
   */
  @Test
  public void testConcurrentOperationsDunitTestOnNonBlockingQueue() throws Exception {
    concurrentOperationsDunitTest(false, Scope.DISTRIBUTED_ACK);
  }

  /**
   * Tests the Non Blocking HARegionQueue by doing concurrent put /remove / take / peek , batch peek
   * operations in multiple regions. The test will have take/remove occuring in all the VMs. This
   * test is targetted to test for hang or exceptions in non blocking queue.
   */
  @Test
  public void testConcurrentOperationsDunitTestOnNonBlockingQueueWithDNoAckRegion()
      throws Exception {
    concurrentOperationsDunitTest(false, Scope.DISTRIBUTED_NO_ACK);
  }

  /**
   * Tests the Blokcing HARegionQueue by doing concurrent put /remove / take / peek , batch peek
   * operations in multiple regions. The test will have take/remove occuring in all the VMs. This
   * test is targetted to test for hang or exceptions in blocking queue.
   */
  @Test
  public void testConcurrentOperationsDunitTestOnBlockingQueue() throws Exception {
    concurrentOperationsDunitTest(true, Scope.DISTRIBUTED_ACK);
  }

  private void concurrentOperationsDunitTest(final boolean createBlockingQueue,
      final Scope rscope) {
    // Create Cache and HARegionQueue in all the 4 VMs.

    CacheSerializableRunnable createRgnsAndQueues = new CacheSerializableRunnable(
        "CreateCache, mirrored Region & HARegionQueue with a CacheListener") {
      @Override
      public void run2() throws CacheException {
        HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
        System.getProperties().put("QueueRemovalThreadWaitTime", "2000");
        cache = test.createCache();
        AttributesFactory factory = new AttributesFactory();
        factory.setScope(rscope);
        factory.setDataPolicy(DataPolicy.REPLICATE);
        HARegionQueueAttributes hrqa = new HARegionQueueAttributes();
        hrqa.setExpiryTime(5);
        try {
          if (createBlockingQueue) {
            hrq = HARegionQueue.getHARegionQueueInstance("testregion1", cache, hrqa,
                HARegionQueue.BLOCKING_HA_QUEUE, false);
          } else {
            hrq = HARegionQueue.getHARegionQueueInstance("testregion1", cache, hrqa,
                HARegionQueue.NON_BLOCKING_HA_QUEUE, false);
          }
        } catch (Exception e) {
          throw new AssertionError(e);
        }
        factory.addCacheListener(new CacheListenerAdapter() {
          @Override
          public void afterCreate(final EntryEvent event) {
            if (toCnt) {
              Conflatable conflatable = new ConflatableObject(event.getKey(), event.getNewValue(),
                  ((EntryEventImpl) event).getEventId(), false, event.getRegion().getFullPath());

              try {
                hrq.put(conflatable);
              } catch (Exception e) {
                fail("The put operation in queue did not succeed due to exception =", e);
              }
            }
          }

          @Override
          public void afterUpdate(final EntryEvent event) {
            if (toCnt) {
              Conflatable conflatable = new ConflatableObject(event.getKey(), event.getNewValue(),
                  ((EntryEventImpl) event).getEventId(), true, event.getRegion().getFullPath());

              try {
                hrq.put(conflatable);
              } catch (Exception e) {
                fail("The put operation in queue did not succeed due to exception =", e);
              }
            }
          }

        });

        cache.createRegion("test_region", factory.create());

      }
    };

    vm0.invoke(createRgnsAndQueues);
    vm1.invoke(createRgnsAndQueues);
    vm2.invoke(createRgnsAndQueues);
    vm3.invoke(createRgnsAndQueues);
    CacheSerializableRunnable spawnThreadsAndperformOps =
        new CacheSerializableRunnable("Spawn multiple threads which do various operations") {

          @Override
          public void run2() throws CacheException {
            opThreads = new RunOp[4 + 2 + 2 + 2];
            for (int i = 0; i < 4; ++i) {
              opThreads[i] = new RunOp(RunOp.PUT, i);
            }
            for (int i = 4; i < 6; ++i) {
              opThreads[i] = new RunOp(RunOp.PEEK, i);
            }

            for (int i = 6; i < 8; ++i) {
              opThreads[i] = new RunOp(RunOp.TAKE, i);
            }

            for (int i = 8; i < 10; ++i) {
              opThreads[i] = new RunOp(RunOp.TAKE, i);
            }

            for (int i = 0; i < opThreads.length; ++i) {
              opThreads[i].start();
            }

          }
        };

    vm0.invokeAsync(spawnThreadsAndperformOps);
    vm1.invokeAsync(spawnThreadsAndperformOps);
    vm2.invokeAsync(spawnThreadsAndperformOps);
    vm3.invokeAsync(spawnThreadsAndperformOps);


    SerializableCallable guaranteeOperationsOccured =
        new SerializableCallable("Check Ops Occurred") {
          @Override
          public Object call() throws CacheException {
            if (opThreads == null) {
              return false;
            }
            for (int i = 0; i < opThreads.length; ++i) {
              if (((RunOp) opThreads[i]).getNumOpsPerformed() == 0) {
                return false;
              }
            }
            return true;
          }
        };

    await()
        .untilAsserted(() -> assertTrue((Boolean) vm0.invoke(guaranteeOperationsOccured)));
    await()
        .untilAsserted(() -> assertTrue((Boolean) vm1.invoke(guaranteeOperationsOccured)));
    await()
        .untilAsserted(() -> assertTrue((Boolean) vm2.invoke(guaranteeOperationsOccured)));
    await()
        .untilAsserted(() -> assertTrue((Boolean) vm3.invoke(guaranteeOperationsOccured)));

    // In case of blocking HARegionQueue do some extra puts so that the
    // blocking threads
    // are exited
    CacheSerializableRunnable toggleFlag =
        new CacheSerializableRunnable("Toggle the flag to signal end of threads") {
          @Override
          public void run2() throws CacheException {
            toCnt = false;
            if (createBlockingQueue) {
              try {
                for (int i = 0; i < 100; ++i) {
                  hrq.put(new ConflatableObject("1", "1", new EventID(new byte[] {1}, 100, i),
                      false, "/x"));
                }
              } catch (Exception e) {
                throw new AssertionError(e);
              }
            }

          }
        };

    vm0.invokeAsync(toggleFlag);
    vm1.invokeAsync(toggleFlag);
    vm2.invokeAsync(toggleFlag);
    vm3.invokeAsync(toggleFlag);

    CacheSerializableRunnable joinWithThreads =
        new CacheSerializableRunnable("Join with the threads") {
          @Override
          public void run2() throws CacheException {
            for (int i = 0; i < opThreads.length; ++i) {

              if (opThreads[i].isInterrupted()) {
                fail("Test failed because  thread encountered exception");
              }
              ThreadUtils.join(opThreads[i], 30 * 1000);
            }
          }
        };

    vm0.invoke(joinWithThreads);
    vm1.invoke(joinWithThreads);
    vm2.invoke(joinWithThreads);
    vm3.invoke(joinWithThreads);
    System.getProperties().remove("QueueRemovalThreadWaitTime");
  }

  /**
   * This is to test the bug which is caused when HARegionQueue object hasnot been fully constructed
   * but as the HARegion has got constructed , it gets visible to QRM Message Thread.
   *
   * TODO: this test runs too long! Shorten run time. 1m 40s on new Mac.
   */
  @Test
  public void testNPEDueToHARegionQueueEscapeInConstructor() {
    // changing EXPIRY_TIME to 5 doesn't change how long the test runs!
    final int EXPIRY_TIME = 30; // test will run for this many seconds
    // Create two HARegionQueue 's in the two VMs. The frequency of QRM thread
    // should be high
    // Check for NullPointeException in the other VM.
    CacheSerializableRunnable createQueuesAndThread =
        new CacheSerializableRunnable("CreateCache, HARegionQueue and start thread") {
          @Override
          public void run2() throws CacheException {
            HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
            // TODO:ASIF: Bcoz of the QRM thread cannot take frequency below
            // 1 second , thus we need to carfully evaluate what to do.
            // For this bug to appear ,without bugfix , qrm needs to run
            // very fast.
            // System.getProperties().put("QueueRemovalThreadWaitTime", new Long(10));
            cache = test.createCache();
            cache.setMessageSyncInterval(1);
            HARegionQueueAttributes hrqa = new HARegionQueueAttributes();
            hrqa.setExpiryTime(EXPIRY_TIME);
            try {
              hrq = HARegionQueue.getHARegionQueueInstance(
                  "testNPEDueToHARegionQueueEscapeInConstructor", cache, hrqa,
                  HARegionQueue.NON_BLOCKING_HA_QUEUE, false);
              // changing OP_COUNT to 20 makes no difference in test time
              final int OP_COUNT = 200;
              // Do 1000 putand 100 take in a separate thread
              for (int i = 0; i < OP_COUNT; ++i) {
                hrq.put(new ConflatableObject(new Long(i), new Long(i),
                    new EventID(new byte[] {0}, 1, i), false, "dummy"));
              }
              opThreads = new Thread[1];
              opThreads[0] = new Thread(new Runnable() {
                @Override
                public void run() {
                  for (int i = 0; i < OP_COUNT; ++i) {
                    try {
                      Object o = hrq.take();
                      if (o == null) {
                        Thread.sleep(50);
                      }
                    } catch (InterruptedException e) {
                      throw new AssertionError(e);
                    }
                  }
                }
              });
              opThreads[0].start();

            } catch (Exception e) {
              throw new AssertionError(e);
            }
          }
        };

    CacheSerializableRunnable createQueues =
        new CacheSerializableRunnable("CreateCache, HARegionQueue ") {
          @Override
          public void run2() throws CacheException {
            createQueuesThread = Thread.currentThread();
            HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
            // System.getProperties().put("QueueRemovalThreadWaitTime",
            // new Long(120000));
            cache = test.createCache();
            cache.setMessageSyncInterval(EXPIRY_TIME);
            HARegionQueueAttributes hrqa = new HARegionQueueAttributes();
            hrqa.setExpiryTime(EXPIRY_TIME);
            try {
              hrq = HARegionQueue.getHARegionQueueInstance(
                  "testNPEDueToHARegionQueueEscapeInConstructor", cache, hrqa,
                  HARegionQueue.NON_BLOCKING_HA_QUEUE, false);
            } catch (Exception e) {
              throw new AssertionError(e);
            }
          }
        };

    CacheSerializableRunnable waitForCreateQueuesThread =
        new CacheSerializableRunnable("joinCreateCache") {
          @Override
          public void run2() {
            WaitCriterion ev = new WaitCriterion() {
              @Override
              public boolean done() {
                return createQueuesThread != null;
              }

              @Override
              public String description() {
                return null;
              }
            };
            GeodeAwaitility.await().untilAsserted(ev);
            join(createQueuesThread, 300 * 1000);
          }
        };

    vm0.invoke(createQueuesAndThread);
    vm1.invokeAsync(createQueues);

    CacheSerializableRunnable joinWithThread =
        new CacheSerializableRunnable("CreateCache, HARegionQueue join with thread") {
          @Override
          public void run2() throws CacheException {
            if (opThreads[0].isInterrupted()) {
              fail("The test has failed as it encountered interrupts in puts & takes");
            }
            ThreadUtils.join(opThreads[0], 30 * 1000);
          }
        };
    vm0.invoke(joinWithThread);
    vm1.invoke(waitForCreateQueuesThread);
  }

  private static class RunOp extends Thread {

    private static final int PUT = 1;
    private static final int TAKE = 2;
    private static final int PEEK = 3;
    private static final int BATCH_PEEK = 4;

    private int opType;
    private int threadID;

    private int numOpsPerformed = 0;

    public RunOp(int opType, int id) {
      super("ID=" + id + ",Op=" + opType);
      this.opType = opType;
      this.threadID = id;
    }

    public int getNumOpsPerformed() {
      return numOpsPerformed;
    }

    @Override
    public void run() {
      Region rgn = cache.getRegion("test_region");
      int counter = 0;
      LogWriter logger = cache.getLogger();
      Conflatable cnf;
      try {
        while (toCnt) {
          // Thread.currentThread().getName() + " before doing operation of
          // type= "+ this.opType);
          switch (opType) {
            case PUT:
              rgn.put("key" + threadID, "val" + counter++);
              if (counter == 10)
                counter = 0;
              break;
            case TAKE:
              cnf = (Conflatable) hrq.take();
              if (logger.fineEnabled() && cnf != null) {
                logger.fine("Object retrieved  by take has key =" + cnf.getKeyToConflate()
                    + " and value as" + cnf.getValueToConflate());
              }
              break;
            case PEEK:
              cnf = (Conflatable) hrq.peek();
              if (logger.fineEnabled() && cnf != null) {
                logger.fine("Object retrieved  by peek has key =" + cnf.getKeyToConflate()
                    + " and value as" + cnf.getValueToConflate());
              }
              // Thread.currentThread().getName() + " before doing remove= "+
              // this.opType);
              hrq.remove();
              break;
            case BATCH_PEEK:
              List confList = hrq.peek(3, 2000);
              if (logger.fineEnabled() && confList != null) {
                logger.fine("Object retrieved  by  batch peek are =" + confList);
              }
              // Thread.currentThread().getName() + " before doing remove= "+
              // this.opType);
              hrq.remove();
              break;

          }
          numOpsPerformed++;
          // Thread.currentThread().getName() + " after Operation of type= "+
          // this.opType);

        }
      } catch (Exception e) {
        throw new AssertionError(e);
      }
    }
  }

  /**
   * Checks the data received by GII, only gets expired after proper construction of HARegionQueue
   * object.
   */
  private static void createHARegionQueueandCheckExpiration() throws Exception {
    HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
    cache = test.createCache();
    HARegionQueueAttributes attrs = new HARegionQueueAttributes();
    attrs.setExpiryTime(1);
    hrq = getHARegionQueueInstance("HARegionQueueDUnitTest_region", cache, attrs,
        NON_BLOCKING_HA_QUEUE, false);
    // wait until we have a dead
    // server
    WaitCriterion ev = new WaitCriterion() {
      @Override
      public boolean done() {
        return hrq.getAvailableIds().size() == 0;
      }

      @Override
      public String description() {
        return null;
      }
    };
    GeodeAwaitility.await().untilAsserted(ev);
    // assertIndexDetailsEquals(0, hrq.getAvailableIds().size());
  }

  @Test
  public void testForDuplicateEvents() throws Exception {
    vm0.invoke(() -> HARegionQueueDUnitTest.createRegionQueue());
    vm1.invoke(() -> HARegionQueueDUnitTest.createRegionQueueandCheckDuplicates());
  }

  /**
   * HARegionQueue should not allow data with duplicate EventIds.
   */
  private static void createRegionQueueandCheckDuplicates() throws Exception {
    HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
    cache = test.createCache();

    hrq = HARegionQueue.getHARegionQueueInstance("HARegionQueueDUnitTest_region", cache,
        HARegionQueue.NON_BLOCKING_HA_QUEUE, false);

    assertEquals(2, hrq.size());

    EventID id1 = new EventID(new byte[] {1}, 1, 1);
    EventID id2 = new EventID(new byte[] {1}, 1, 2);
    ConflatableObject c1 =
        new ConflatableObject("1", "1", id1, false, "HARegionQueueDUnitTest_region");
    ConflatableObject c2 =
        new ConflatableObject("2", "2", id2, false, "HARegionQueueDUnitTest_region");

    hrq.put(c1);
    hrq.put(c2);

    // HARegion size should be 2 as data with same EventIDs is inserted into the queue
    assertEquals(2, hrq.size());
  }
}