/*
 * 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 org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.internal.lang.SystemPropertyHelper.HA_REGION_QUEUE_EXPIRY_TIME_PROPERTY;
import static org.apache.geode.internal.statistics.StatisticsClockFactory.disabledClock;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.not;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.CoreMatchers.sameInstance;
import static org.hamcrest.number.OrderingComparison.greaterThanOrEqualTo;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

import org.awaitility.core.ThrowingRunnable;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.contrib.java.lang.system.RestoreSystemProperties;
import org.junit.experimental.categories.Category;
import org.junit.rules.ErrorCollector;
import org.junit.rules.TestName;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl;
import org.apache.geode.internal.cache.tier.sockets.HAEventWrapper;
import org.apache.geode.internal.lang.SystemProperty;
import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

/**
 * This is a test for the APIs of a HARegionQueue and verifies that the head, tail and size counters
 * are updated properly.
 *
 * <p>
 * TODO: use ExecutorServiceRule instead of raw threads (move CyclicBarrier into rule if needed)
 */
@Category({ClientSubscriptionTest.class})
public class HARegionQueueJUnitTest {

  /** total number of threads doing put operations */
  private static final int TOTAL_PUT_THREADS = 10;

  private static HARegionQueue hrqForTestSafeConflationRemoval;
  private static List list1;

  protected InternalCache cache;
  private HARegionQueue haRegionQueue;

  @Rule
  public RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();

  @Rule
  public ErrorCollector errorCollector = new ErrorCollector();

  @Rule
  public TestName testName = new TestName();

  @Before
  public void setUp() throws Exception {
    cache = createCache();
  }

  @After
  public void tearDown() throws Exception {
    cache.close();
    hrqForTestSafeConflationRemoval = null;
  }

  /**
   * This test does the following :<br>
   * 1)Create producer threads to add objects to queue <br>
   * 2)Start all the producers so that they complete their puts <br>
   * 3)Wait till all put-threads complete their job <br>
   * 4)verify that the size of the queue is equal to the total number of puts done by all producers
   */
  @Test
  public void testQueuePutWithoutConflation() throws Exception {
    haRegionQueue = createHARegionQueue(testName.getMethodName());
    int putPerProducer = 20;
    createAndRunProducers(false, false, false, putPerProducer);
    assertThat(haRegionQueue.size(), is(putPerProducer * TOTAL_PUT_THREADS));
  }

  /**
   * This test does the following :<br>
   * 1)Create producer threads to add objects to queue <br>
   * 2)Start all the producers,all of them will do puts against same set of keys <br>
   * 3)Wait till all put-threads complete their job <br>
   * 4)verify that the size of the queue is equal to the total number of puts done by one thread (as
   * rest of them will conflate)
   */
  @Test
  public void testQueuePutWithConflation() throws Exception {
    haRegionQueue = createHARegionQueue(testName.getMethodName());
    int putPerProducer = 20;
    createAndRunProducers(true, false, true, putPerProducer);
    assertThat(haRegionQueue.size(), is(putPerProducer));
  }

  /**
   * This test does the following :<br>
   * 1)Create producer threads to add objects to queue <br>
   * 2)Start all the producers,all of them will do puts against same set of ids <br>
   * 3)Wait till all put-threads complete their job <br>
   * 4)verify that the size of the queue is equal to the total number of puts done by one thread (as
   * rest of them will be duplicates and hence will be replaced)
   */
  @Test
  public void testQueuePutWithDuplicates() throws Exception {
    haRegionQueue = createHARegionQueue(testName.getMethodName());
    int putPerProducer = 20;
    createAndRunProducers(false, false, true, putPerProducer);
    assertThat(haRegionQueue.size(), is(putPerProducer * TOTAL_PUT_THREADS));
  }

  /*
   * Test method for 'org.apache.geode.internal.cache.ha.HARegionQueue.addDispatchedMessage(Object)'
   */
  @Test
  public void testAddDispatchedMessageObject() throws Exception {
    haRegionQueue = createHARegionQueue(testName.getMethodName());
    assertThat(HARegionQueue.getDispatchedMessagesMapForTesting().isEmpty(), is(true));

    haRegionQueue.addDispatchedMessage(new ThreadIdentifier(new byte[1], 1), 1);
    haRegionQueue.addDispatchedMessage(new ThreadIdentifier(new byte[1], 2), 2);

    assertThat(!HARegionQueue.getDispatchedMessagesMapForTesting().isEmpty(), is(true));
  }

  /**
   * tests the blocking peek functionality of BlockingHARegionQueue
   */
  @Test
  public void testBlockQueue() throws Exception {
    HARegionQueue regionQueue = HARegionQueue.getHARegionQueueInstance(testName.getMethodName(),
        cache, HARegionQueue.BLOCKING_HA_QUEUE, false, disabledClock());
    Thread[] threads = new Thread[10];
    int threadsLength = threads.length;
    CyclicBarrier barrier = new CyclicBarrier(threadsLength + 1);

    for (int i = 0; i < threadsLength; i++) {
      threads[i] = new Thread() {
        @Override
        public void run() {
          try {
            barrier.await();
            long startTime = System.currentTimeMillis();
            Object obj = regionQueue.peek();
            if (obj == null) {
              errorCollector.addError(new AssertionError(
                  "Failed :  failed since object was null and was not expected to be null"));
            }
            long totalTime = System.currentTimeMillis() - startTime;

            if (totalTime < 2000) {
              errorCollector.addError(new AssertionError(
                  " Failed :  Expected time to be greater than 2000 but it is not so "));
            }
          } catch (Exception e) {
            errorCollector.addError(e);
          }
        }
      };
    }

    for (Thread thread1 : threads) {
      thread1.start();
    }

    barrier.await();

    Thread.sleep(5000);

    EventID id = new EventID(new byte[] {1}, 1, 1);
    regionQueue.put(new ConflatableObject("key", "value", id, false, testName.getMethodName()));

    long startTime = System.currentTimeMillis();
    for (Thread thread : threads) {
      ThreadUtils.join(thread, 60 * 1000);
    }

    long totalTime = System.currentTimeMillis() - startTime;

    if (totalTime >= 60000) {
      fail(" Test taken too long ");
    }
  }

  /**
   * tests whether expiry of entry in the region queue occurs as expected
   */
  @Test
  public void testExpiryPositive() throws Exception {
    HARegionQueueAttributes haa = new HARegionQueueAttributes();
    haa.setExpiryTime(1);

    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName(), haa);
    long start = System.currentTimeMillis();

    regionQueue.put(new ConflatableObject("key", "value", new EventID(new byte[] {1}, 1, 1), true,
        testName.getMethodName()));

    Map map = (Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName());
    waitAtLeast(1000, start, () -> {
      assertThat(map, is(Collections.emptyMap()));
      assertThat(regionQueue.getRegion().keys(), is(Collections.emptySet()));
    });
  }

  /**
   * tests whether expiry of a conflated entry in the region queue occurs as expected
   */
  @Test
  public void testExpiryPositiveWithConflation() throws Exception {
    HARegionQueueAttributes haa = new HARegionQueueAttributes();
    haa.setExpiryTime(1);

    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName(), haa);
    long start = System.currentTimeMillis();

    regionQueue.put(new ConflatableObject("key", "value", new EventID(new byte[] {1}, 1, 1), true,
        testName.getMethodName()));

    regionQueue.put(new ConflatableObject("key", "newValue", new EventID(new byte[] {1}, 1, 2),
        true, testName.getMethodName()));

    assertThat(
        " Expected region size not to be zero since expiry time has not been exceeded but it is not so ",
        !regionQueue.isEmpty(), is(true));
    assertThat(
        " Expected the available id's size not  to be zero since expiry time has not  been exceeded but it is not so ",
        !regionQueue.getAvailableIds().isEmpty(), is(true));
    assertThat(
        " Expected conflation map size not  to be zero since expiry time has not been exceeded but it is not so "
            + ((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName()))
                .get("key"),
        ((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName())).get("key"),
        not(sameInstance(null)));
    assertThat(
        " Expected eventID map size not to be zero since expiry time has not been exceeded but it is not so ",
        !regionQueue.getEventsMapForTesting().isEmpty(), is(true));

    waitAtLeast(1000, start, () -> {
      assertThat(regionQueue.getRegion().keys(), is(Collections.emptySet()));
      assertThat(regionQueue.getAvailableIds(), is(Collections.emptySet()));
      assertThat(regionQueue.getConflationMapForTesting().get(testName.getMethodName()),
          is(Collections.emptyMap()));
      assertThat(regionQueue.getEventsMapForTesting(), is(Collections.emptyMap()));
    });
  }

  /**
   * tests a ThreadId not being expired if it was updated
   */
  @Test
  public void testNoExpiryOfThreadId() throws Exception {
    HARegionQueueAttributes haa = new HARegionQueueAttributes();
    haa.setExpiryTime(45);

    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName(), haa);
    EventID ev1 = new EventID(new byte[] {1}, 1, 1);
    EventID ev2 = new EventID(new byte[] {1}, 1, 2);
    Conflatable cf1 = new ConflatableObject("key", "value", ev1, true, testName.getMethodName());
    Conflatable cf2 = new ConflatableObject("key", "value2", ev2, true, testName.getMethodName());

    regionQueue.put(cf1);
    long tailKey = regionQueue.tailKey.get();
    regionQueue.put(cf2);

    // Invalidate will trigger the expiration of the entry
    // See HARegionQueue.createCacheListenerForHARegion
    regionQueue.getRegion().invalidate(tailKey);

    assertThat(
        " Expected region size not to be zero since expiry time has not been exceeded but it is not so ",
        !regionQueue.isEmpty(), is(true));
    assertThat(" Expected the available id's size not  to have counter 1 but it has ",
        !regionQueue.getAvailableIds().contains(1L), is(true));
    assertThat(" Expected the available id's size to have counter 2 but it does not have ",
        regionQueue.getAvailableIds().contains(2L), is(true));
    assertThat(" Expected eventID map not to have the first event, but it has",
        !regionQueue.getCurrentCounterSet(ev1).contains(1L), is(true));
    assertThat(" Expected eventID map to have the second event, but it does not",
        regionQueue.getCurrentCounterSet(ev2).contains(2L), is(true));
  }

  /**
   * Tests a QueueRemovalMessage coming before a localPut(). The localPut() should result in no data
   * being put in the queue
   */
  @Test
  public void testQRMComingBeforeLocalPut() throws Exception {
    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName());
    EventID id = new EventID(new byte[] {1}, 1, 1);

    regionQueue.removeDispatchedEvents(id);
    regionQueue.put(new ConflatableObject("key", "value", id, true, testName.getMethodName()));

    assertThat(" Expected key to be null since QRM for the message id had already arrived ",
        !regionQueue.getRegion().containsKey(1L), is(true));
  }

  /**
   * test verifies correct expiry of ThreadIdentifier in the HARQ if no corresponding put comes
   */
  @Test
  public void testOnlyQRMComing() throws Exception {
    HARegionQueueAttributes harqAttr = new HARegionQueueAttributes();
    harqAttr.setExpiryTime(1);

    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName(), harqAttr);
    EventID id = new EventID(new byte[] {1}, 1, 1);
    long start = System.currentTimeMillis();

    regionQueue.removeDispatchedEvents(id);

    assertThat(" Expected testingID to be present since only QRM achieved ",
        regionQueue.getRegion().containsKey(new ThreadIdentifier(new byte[] {1}, 1)), is(true));

    waitAtLeast(1000, start,
        () -> assertThat(
            " Expected testingID not to be present since it should have expired after 2.5 seconds",
            !regionQueue.getRegion().containsKey(new ThreadIdentifier(new byte[] {1}, 1)),
            is(true)));
  }

  /**
   * test all relevant data structures are updated on a local put
   */
  @Test
  public void testPutPath() throws Exception {
    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName());
    Conflatable cf = new ConflatableObject("key", "value", new EventID(new byte[] {1}, 1, 1), true,
        testName.getMethodName());

    regionQueue.put(cf);

    assertThat(" Expected region peek to return cf but it is not so ", regionQueue.peek(), is(cf));
    assertThat(
        " Expected the available id's size not  to be zero since expiry time has not  been exceeded but it is not so ",
        !regionQueue.getAvailableIds().isEmpty(), is(true));
    assertThat(
        " Expected conflation map to have entry for this key since expiry time has not been exceeded but it is not so ",
        ((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName())).get("key"),
        is(1L));
    assertThat(
        " Expected eventID map size not to be zero since expiry time has not been exceeded but it is not so ",
        !regionQueue.getEventsMapForTesting().isEmpty(), is(true));
  }

  /**
   * - adds 10 items - sets last dispatched as 5th - verify no data pertaining to the first five is
   * there - verify the next five entries and their relevant data is present
   */
  @Test
  public void testQRMDispatch() throws Exception {
    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName());
    Conflatable[] cf = new Conflatable[10];

    // put 10 conflatable objects
    for (int i = 0; i < 10; i++) {
      cf[i] = new ConflatableObject("key" + i, "value", new EventID(new byte[] {1}, 1, i), true,
          testName.getMethodName());
      regionQueue.put(cf[i]);
    }

    // remove the first 5 by giving the right sequence id
    regionQueue.removeDispatchedEvents(new EventID(new byte[] {1}, 1, 4));

    // verify 1-5 not in region
    for (int i = 1; i < 6; i++) {
      assertThat(!regionQueue.getRegion().containsKey((long) i), is(true));
    }

    // verify 6-10 still in region queue
    for (int i = 6; i < 11; i++) {
      assertThat(regionQueue.getRegion().containsKey((long) i), is(true));
    }

    // verify 1-5 not in conflation map
    for (int i = 0; i < 5; i++) {
      assertThat(!((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName()))
          .containsKey("key" + i), is(true));
    }

    // verify 6-10 in conflation map
    for (int i = 5; i < 10; i++) {
      assertThat(((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName()))
          .containsKey("key" + i), is(true));
    }

    EventID eid = new EventID(new byte[] {1}, 1, 6);

    // verify 1-5 not in eventMap
    for (int i = 1; i < 6; i++) {
      assertThat(!regionQueue.getCurrentCounterSet(eid).contains((long) i), is(true));
    }

    // verify 6-10 in event Map
    for (int i = 6; i < 11; i++) {
      assertThat(regionQueue.getCurrentCounterSet(eid).contains((long) i), is(true));
    }

    // verify 1-5 not in available Id's map
    for (int i = 1; i < 6; i++) {
      assertThat(!regionQueue.getAvailableIds().contains((long) i), is(true));
    }

    // verify 6-10 in available id's map
    for (int i = 6; i < 11; i++) {
      assertThat(regionQueue.getAvailableIds().contains((long) i), is(true));
    }
  }

  /**
   * - send Dispatch message for sequence id 7 - put from sequence id 1 - id 10 - verify data for
   * 1-7 not there - verify data for 8-10 is there
   */
  @Test
  public void testQRMBeforePut() throws Exception {
    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName());

    EventID[] ids = new EventID[10];

    for (int i = 0; i < 10; i++) {
      ids[i] = new EventID(new byte[] {1}, 1, i);
    }

    // first get the qrm message for the seventh id
    regionQueue.removeDispatchedEvents(ids[6]);
    Conflatable[] cf = new Conflatable[10];

    // put 10 conflatable objects
    for (int i = 0; i < 10; i++) {
      cf[i] = new ConflatableObject("key" + i, "value", ids[i], true, testName.getMethodName());
      regionQueue.put(cf[i]);
    }

    // verify 1-7 not in region
    Set values = (Set) regionQueue.getRegion().values();

    for (int i = 0; i < 7; i++) {
      System.out.println(i);
      assertThat(!values.contains(cf[i]), is(true));
    }

    // verify 8-10 still in region queue
    for (int i = 7; i < 10; i++) {
      System.out.println(i);
      assertThat(values.contains(cf[i]), is(true));
    }

    // verify 1-8 not in conflation map
    for (int i = 0; i < 7; i++) {
      assertThat(!((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName()))
          .containsKey("key" + i), is(true));
    }

    // verify 8-10 in conflation map
    for (int i = 7; i < 10; i++) {
      assertThat(((Map) regionQueue.getConflationMapForTesting().get(testName.getMethodName()))
          .containsKey("key" + i), is(true));
    }

    EventID eid = new EventID(new byte[] {1}, 1, 6);

    // verify 1-7 not in eventMap
    for (int i = 4; i < 11; i++) {
      assertThat(!regionQueue.getCurrentCounterSet(eid).contains((long) i), is(true));
    }

    // verify 8-10 in event Map
    for (int i = 1; i < 4; i++) {
      assertThat(regionQueue.getCurrentCounterSet(eid).contains((long) i), is(true));
    }

    // verify 1-7 not in available Id's map
    for (int i = 4; i < 11; i++) {
      assertThat(!regionQueue.getAvailableIds().contains((long) i), is(true));
    }

    // verify 8-10 in available id's map
    for (int i = 1; i < 4; i++) {
      assertThat(regionQueue.getAvailableIds().contains((long) i), is(true));
    }
  }

  /**
   * test to verify conflation happens as expected
   */
  @Test
  public void testConflation() throws Exception {
    HARegionQueue regionQueue = createHARegionQueue(testName.getMethodName());
    EventID ev1 = new EventID(new byte[] {1}, 1, 1);
    EventID ev2 = new EventID(new byte[] {1}, 2, 2);
    Conflatable cf1 = new ConflatableObject("key", "value", ev1, true, testName.getMethodName());
    Conflatable cf2 = new ConflatableObject("key", "value2", ev2, true, testName.getMethodName());
    regionQueue.put(cf1);

    Map conflationMap = regionQueue.getConflationMapForTesting();
    assertThat(((Map) conflationMap.get(testName.getMethodName())).get("key"), is(1L));

    regionQueue.put(cf2);

    // verify the conflation map has recorded the new key
    assertThat(((Map) conflationMap.get(testName.getMethodName())).get("key"), is(2L));
    // the old key should not be present
    assertThat(!regionQueue.getRegion().containsKey(1L), is(true));
    // available ids should not contain the old id (the old position)
    assertThat(!regionQueue.getAvailableIds().contains(1L), is(true));
    // available id should have the new id (the new position)
    assertThat(regionQueue.getAvailableIds().contains(2L), is(true));
    // events map should not contain the old position
    assertThat(regionQueue.getCurrentCounterSet(ev1).isEmpty(), is(true));
    // events map should contain the new position
    assertThat(regionQueue.getCurrentCounterSet(ev2).contains(2L), is(true));
  }

  /**
   * Tests whether the QRM message removes the events correctly from the DACE & Conflation Map. The
   * events which are of ID greater than that contained in QRM should stay
   */
  @Test
  public void testQRM() throws Exception {
    RegionQueue regionqueue = createHARegionQueue(testName.getMethodName());

    for (int i = 0; i < 10; ++i) {
      regionqueue.put(new ConflatableObject("key" + (i + 1), "value",
          new EventID(new byte[] {1}, 1, i + 1), true, testName.getMethodName()));
    }

    EventID qrmID = new EventID(new byte[] {1}, 1, 5);
    ((HARegionQueue) regionqueue).removeDispatchedEvents(qrmID);
    Map conflationMap = ((HARegionQueue) regionqueue).getConflationMapForTesting();
    assertThat(((Map) conflationMap.get(testName.getMethodName())).size(), is(5));

    Set availableIDs = ((HARegionQueue) regionqueue).getAvailableIds();
    Set counters = ((HARegionQueue) regionqueue).getCurrentCounterSet(qrmID);

    assertThat(availableIDs.size(), is(5));
    assertThat(counters.size(), is(5));

    for (int i = 5; i < 10; ++i) {
      assertThat(((Map) (conflationMap.get(testName.getMethodName()))).containsKey("key" + (i + 1)),
          is(true));
      assertThat(availableIDs.contains((long) (i + 1)), is(true));
      assertThat(counters.contains((long) (i + 1)), is(true));
    }

    Region rgn = ((HARegionQueue) regionqueue).getRegion();
    assertThat(rgn.keySet().size(), is(6));
  }

  /**
   * This test tests safe removal from the conflation map. i.e operations should only remove old
   * values and not the latest value
   */
  @Test
  public void testSafeConflationRemoval() throws Exception {
    hrqForTestSafeConflationRemoval = new HARQTestClass("testSafeConflationRemoval", cache);
    Conflatable cf1 = new ConflatableObject("key1", "value", new EventID(new byte[] {1}, 1, 1),
        true, "testSafeConflationRemoval");

    hrqForTestSafeConflationRemoval.put(cf1);
    hrqForTestSafeConflationRemoval.removeDispatchedEvents(new EventID(new byte[] {1}, 1, 1));

    Map map = (Map) hrqForTestSafeConflationRemoval.getConflationMapForTesting()
        .get("testSafeConflationRemoval");

    assertThat(
        "Expected the counter to be 2 since it should not have been deleted but it is not so ",
        map.get("key1"), is(2L));
  }

  /**
   * This test tests remove operation is causing the insertion of sequence ID for existing
   * ThreadIdentifier object and concurrently the QRM thread is iterating over the Map to form the
   * Data Set for dispatch. There should not be any Data Loss
   *
   * In this test, first we add x number of events for unique thread id for the same region then we
   * start two concurrent threads. One which does add to dispatched events map with sequence id's
   * greater than x but the same ThreadIds as previous. The second thread does createMessageList
   * (which removes from the dispatched events map) and stores the list
   *
   * After the two threads have operated, createMessageList is called again and the data is stored
   * in another list.
   *
   * The data in the list is populated on a map against the threadId.
   *
   * It is then verified to see that all the sequence should be greater than x
   */
  @Test
  public void testConcurrentDispatcherAndRemovalForSameRegionSameThreadId() throws Exception {
    long numberOfIterations = 1000;
    HARegionQueue hrq = createHARegionQueue(testName.getMethodName());
    HARegionQueue.stopQRMThread();
    ThreadIdentifier[] ids = new ThreadIdentifier[(int) numberOfIterations];

    for (int i = 0; i < numberOfIterations; i++) {
      ids[i] = new ThreadIdentifier(new byte[] {1}, i);
      hrq.addDispatchedMessage(ids[i], i);
    }

    Thread thread1 = new Thread() {
      @Override
      public void run() {
        try {
          Thread.sleep(600);
        } catch (InterruptedException e) {
          errorCollector.addError(e);
        }
        list1 = HARegionQueue.createMessageListForTesting();
      }
    };

    Thread thread2 = new Thread() {
      @Override
      public void run() {
        try {
          Thread.sleep(480);
        } catch (InterruptedException e) {
          errorCollector.addError(e);
        }
        for (int i = 0; i < numberOfIterations; i++) {
          hrq.addDispatchedMessage(ids[i], i + numberOfIterations);
        }
      }
    };

    thread1.start();
    thread2.start();
    ThreadUtils.join(thread1, 30 * 1000);
    ThreadUtils.join(thread2, 30 * 1000);
    List list2 = HARegionQueue.createMessageListForTesting();
    Iterator iterator = list1.iterator();
    boolean doOnce = false;
    EventID id;
    Map map = new HashMap();

    while (iterator.hasNext()) {
      if (!doOnce) {
        iterator.next();
        iterator.next();
        doOnce = true;
      } else {
        id = (EventID) iterator.next();
        map.put(id.getThreadID(), id.getSequenceID());
      }
    }

    iterator = list2.iterator();
    doOnce = false;

    while (iterator.hasNext()) {
      if (!doOnce) {
        iterator.next();
        iterator.next();
        doOnce = true;
      } else {
        id = (EventID) iterator.next();
        map.put(id.getThreadID(), id.getSequenceID());
      }
    }

    iterator = map.values().iterator();
    Long max = numberOfIterations;
    while (iterator.hasNext()) {
      Long next = (Long) iterator.next();
      assertThat(
          " Expected all the sequence ID's to be greater than " + max
              + " but it is not so. Got sequence id " + next,
          next.compareTo(max), greaterThanOrEqualTo(0));
    }
  }

  /**
   * This test remove operation is updating the sequence ID for a ThreadIdentifier and concurrently
   * the QRM thread is iterating over the Map to form the Data Set. There should not be any DataLoss
   *
   * In this test, first we add x number of events for unique thread id for the same region then we
   * start two concurrent threads. One which does add to dispatched events map with sequence id's
   * greater than x and the second one which does createMessageList (which removes from the
   * dispatched events map) and stores the list
   *
   * After the two threads have operated, createMessageList is called again and the data is stored
   * in another list.
   *
   * The data in the list is populated on a map against the threadId.
   *
   * It is then verified to see that the map size should be 2x
   */
  @Test
  public void testConcurrentDispatcherAndRemovalForSameRegionDifferentThreadId() throws Exception {
    int numberOfIterations = 1000;
    HARegionQueue hrq = createHARegionQueue(testName.getMethodName());
    HARegionQueue.stopQRMThread();
    ThreadIdentifier[] ids = new ThreadIdentifier[(int) numberOfIterations];

    for (int i = 0; i < numberOfIterations; i++) {
      ids[i] = new ThreadIdentifier(new byte[] {1}, i);
      hrq.addDispatchedMessage(ids[i], i);
    }

    Thread thread1 = new Thread() {
      @Override
      public void run() {
        try {
          Thread.sleep(600);
        } catch (InterruptedException e) {
          errorCollector.addError(e);
        }
        list1 = HARegionQueue.createMessageListForTesting();
      }
    };

    Thread thread2 = new Thread() {
      @Override
      public void run() {
        try {
          Thread.sleep(480);
        } catch (InterruptedException e) {
          errorCollector.addError(e);
        }
        for (int i = 0; i < numberOfIterations; i++) {
          ids[i] = new ThreadIdentifier(new byte[] {1}, i + numberOfIterations);
          hrq.addDispatchedMessage(ids[i], i + numberOfIterations);
        }
      }
    };

    thread1.start();
    thread2.start();
    ThreadUtils.join(thread1, 30 * 1000);
    ThreadUtils.join(thread2, 30 * 1000);
    List list2 = HARegionQueue.createMessageListForTesting();
    Iterator iterator = list1.iterator();
    boolean doOnce = false;
    EventID id;
    Map map = new HashMap();

    while (iterator.hasNext()) {
      if (!doOnce) {
        iterator.next();
        iterator.next();
        doOnce = true;
      } else {
        id = (EventID) iterator.next();
        map.put(id.getThreadID(), id.getSequenceID());
      }
    }

    iterator = list2.iterator();
    doOnce = false;

    while (iterator.hasNext()) {
      if (!doOnce) {
        iterator.next();
        iterator.next();
        doOnce = true;
      } else {
        id = (EventID) iterator.next();
        map.put(id.getThreadID(), id.getSequenceID());
      }
    }
    assertThat(
        " Expected the map size to be " + 2 * numberOfIterations + " but it is " + map.size(),
        map.size(), is(2 * numberOfIterations));
  }

  /**
   * This test tests remove operation is causing the insertion of a sequence ID for a new HA Region
   * and concurrently the QRM thread is iterating over the Map to form the Data Set for dispatch.
   * There should not be any Data Loss
   *
   * In this test, first we add x number of events for unique thread id for the 2 regions then we
   * start two concurrent threads. One which does add to dispatched events map to 3 new regions
   *
   * the second thread does createMessageList (which removes from the dispatched events map) and
   * stores the list
   *
   * After the two threads have operated, createMessageList is called again and the data is stored
   * in another list.
   *
   * The data in the list is populated on a map against the threadId.
   *
   * It is then verified to see that a total of x entries are present in the map
   */
  @Test
  public void testConcurrentDispatcherAndRemovalForMultipleRegionsSameThreadId() throws Exception {
    int numberOfIterations = 10000;
    HARegionQueue hrq1 = createHARegionQueue(testName.getMethodName() + "-1");
    HARegionQueue hrq2 = createHARegionQueue(testName.getMethodName() + "-2");
    HARegionQueue hrq3 = createHARegionQueue(testName.getMethodName() + "-3");
    HARegionQueue hrq4 = createHARegionQueue(testName.getMethodName() + "-4");
    HARegionQueue hrq5 = createHARegionQueue(testName.getMethodName() + "-5");

    HARegionQueue.stopQRMThread();

    ThreadIdentifier[] ids = new ThreadIdentifier[(int) numberOfIterations];

