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

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Declarable;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.DiskStoreFactory;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.FixedPartitionAttributes;
import org.apache.geode.cache.FixedPartitionResolver;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueStats;
import org.apache.geode.cache.client.internal.LocatorDiscoveryCallbackAdapter;
import org.apache.geode.cache.control.RebalanceFactory;
import org.apache.geode.cache.control.RebalanceOperation;
import org.apache.geode.cache.control.RebalanceResults;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.wan.GatewayEventFilter;
import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter;
import org.apache.geode.cache.wan.GatewayReceiver;
import org.apache.geode.cache.wan.GatewayReceiverFactory;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
import org.apache.geode.cache.wan.GatewaySenderFactory;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalLocator;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserver;
import org.apache.geode.internal.size.Sizeable;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Assert;
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.VM;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.dunit.rules.ClusterStartupRule;
import org.apache.geode.test.dunit.rules.DistributedRule;


/**
 * @deprecated Please use {@link DistributedRule} and Geode User APIs or {@link ClusterStartupRule}
 *             instead.
 */
@Deprecated
public class AsyncEventQueueTestBase extends JUnit4DistributedTestCase {

  protected static Cache cache;

  protected static VM vm0;

  protected static VM vm1;

  protected static VM vm2;

  protected static VM vm3;

  protected static VM vm4;

  protected static AsyncEventListener eventListener1;

  private static final long MAX_WAIT = 60000;

  protected static GatewayEventFilter eventFilter;

  protected static boolean destroyFlag = false;

  protected static List<Integer> dispatcherThreads = new ArrayList<>(Arrays.asList(1, 3, 5));

  // this will be set for each test method run with one of the values from above
  // list
  protected static int numDispatcherThreadsForTheRun = 1;

  public AsyncEventQueueTestBase() {
    super();
  }

  @Override
  public final void preSetUp() throws Exception {
    final Host host = Host.getHost(0);
    vm0 = host.getVM(0);
    vm1 = host.getVM(1);
    vm2 = host.getVM(2);
    vm3 = host.getVM(3);
    vm4 = host.getVM(4);
  }

  @Override
  public final void postSetUp() throws Exception {
    // this is done to vary the number of dispatchers for sender
    // during every test method run
    shuffleNumDispatcherThreads();
    Invoke.invokeInEveryVM(() -> setNumDispatcherThreadsForTheRun(dispatcherThreads.get(0)));
  }

  public static void shuffleNumDispatcherThreads() {
    Collections.shuffle(dispatcherThreads);
  }

  public static void setNumDispatcherThreadsForTheRun(int numThreads) {
    numDispatcherThreadsForTheRun = numThreads;
  }

  public static Integer createFirstLocatorWithDSId(int dsId) {
    if (Locator.hasLocator()) {
      Locator.getLocator().stop();
    }
    AsyncEventQueueTestBase test = new AsyncEventQueueTestBase();
    int port = AvailablePortHelper.getRandomAvailableTCPPort();
    Properties props = test.getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "0");
    // props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId);
    props.setProperty(LOCATORS, "localhost[" + port + "]");
    props.setProperty(START_LOCATOR,
        "localhost[" + port + "],server=true,peer=true,hostname-for-clients=localhost");
    test.startLocatorDistributedSystem(props);
    return port;
  }

  public static Integer createFirstRemoteLocator(int dsId, int remoteLocPort) {
    AsyncEventQueueTestBase test = new AsyncEventQueueTestBase();
    int port = AvailablePortHelper.getRandomAvailableTCPPort();
    Properties props = test.getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId);
    props.setProperty(LOCATORS, "localhost[" + port + "]");
    props.setProperty(START_LOCATOR,
        "localhost[" + port + "],server=true,peer=true,hostname-for-clients=localhost");
    props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocPort + "]");
    test.startLocatorDistributedSystem(props);
    return port;
  }

  private void startLocatorDistributedSystem(Properties props) {
    // Start start the locator with a LOCATOR_DM_TYPE and not a NORMAL_DM_TYPE
    System.setProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE, "true");
    try {
      getSystem(props);
    } finally {
      System.clearProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE);
    }
  }

  public static void createReplicatedRegionWithAsyncEventQueue(String regionName,
      String asyncQueueIds, Boolean offHeap) {
    IgnoredException exp1 =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());
    try {
      AttributesFactory fact = new AttributesFactory();
      addAsyncEventQueueIds(fact, asyncQueueIds);
      fact.setDataPolicy(DataPolicy.REPLICATE);
      fact.setOffHeap(offHeap);
      RegionFactory regionFactory = cache.createRegionFactory(fact.create());
      Region r = regionFactory.create(regionName);
      assertNotNull(r);
    } finally {
      exp1.remove();
    }
  }

  public static void createReplicatedRegionWithCacheLoaderAndAsyncEventQueue(String regionName,
      String asyncQueueIds) {

    AttributesFactory fact = new AttributesFactory();
    addAsyncEventQueueIds(fact, asyncQueueIds);
    fact.setDataPolicy(DataPolicy.REPLICATE);
    // set the CacheLoader
    fact.setCacheLoader(new MyCacheLoader());
    RegionFactory regionFactory = cache.createRegionFactory(fact.create());
    Region r = regionFactory.create(regionName);
    assertNotNull(r);
  }

  private static void addAsyncEventQueueIds(AttributesFactory fact, String asyncQueueIds) {
    if (asyncQueueIds != null) {
      StringTokenizer tokenizer = new StringTokenizer(asyncQueueIds, ",");
      while (tokenizer.hasMoreTokens()) {
        String asyncQueueId = tokenizer.nextToken();
        fact.addAsyncEventQueueId(asyncQueueId);
      }
    }
  }

  public static void createReplicatedRegionWithSenderAndAsyncEventQueue(String regionName,
      String senderIds, String asyncChannelId, Boolean offHeap) {
    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());
    try {

      AttributesFactory fact = new AttributesFactory();
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          fact.addGatewaySenderId(senderId);
        }
      }
      fact.setDataPolicy(DataPolicy.REPLICATE);
      fact.setOffHeap(offHeap);
      fact.setScope(Scope.DISTRIBUTED_ACK);
      RegionFactory regionFactory = cache.createRegionFactory(fact.create());
      regionFactory.addAsyncEventQueueId(asyncChannelId);
      Region r = regionFactory.create(regionName);
      assertNotNull(r);
    } finally {
      exp.remove();
    }
  }

  public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous) {
    createAsyncEventQueue(asyncChannelId, isParallel, maxMemory, batchSize, isConflation,
        isPersistent, diskStoreName, isDiskSynchronous, new MyAsyncEventListener());
  }

  public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous,
      final AsyncEventListener asyncEventListener) {
    createAsyncEventQueue(asyncChannelId, isParallel, maxMemory, batchSize, isConflation,
        isPersistent, diskStoreName, isDiskSynchronous, numDispatcherThreadsForTheRun,
        asyncEventListener);
  }

  public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous, int numDispatcherThreads,
      final AsyncEventListener asyncEventListener) {
    createDiskStore(asyncChannelId, diskStoreName);

    AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory,
        batchSize, isPersistent, diskStoreName);
    factory.setDiskSynchronous(isDiskSynchronous);
    factory.setBatchConflationEnabled(isConflation);
    // set dispatcher threads
    factory.setDispatcherThreads(numDispatcherThreads);
    // Set GatewayEventSubstitutionFilter
    AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener);
  }

  private static void createDiskStore(String asyncChannelId, String diskStoreName) {
    if (diskStoreName != null) {
      File directory = new File(
          asyncChannelId + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum());
      directory.mkdir();
      File[] dirs1 = new File[] {directory};
      DiskStoreFactory dsf = cache.createDiskStoreFactory();
      dsf.setDiskDirs(dirs1);
      DiskStore ds = dsf.create(diskStoreName);
    }
  }

  public static void createAsyncEventQueueWithListener2(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isPersistent, String diskStoreName) {

    createDiskStore(asyncChannelId, diskStoreName);

    AsyncEventListener asyncEventListener = new MyAsyncEventListener2();

    AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory,
        batchSize, isPersistent, diskStoreName);
    // set dispatcher threads
    factory.setDispatcherThreads(numDispatcherThreadsForTheRun);
    AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener);
  }

  public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous, String asyncListenerClass) throws Exception {
    createAsyncEventQueue(asyncChannelId, isParallel, maxMemory, batchSize, isConflation,
        isPersistent, diskStoreName, isDiskSynchronous, asyncListenerClass, null);
  }

  public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous, String asyncListenerClass,
      String substitutionFilterClass) throws Exception {

    createDiskStore(asyncChannelId, diskStoreName);

    AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory,
        batchSize, isPersistent, diskStoreName);
    factory.setDiskSynchronous(isDiskSynchronous);
    factory.setBatchConflationEnabled(isConflation);
    if (substitutionFilterClass != null) {
      factory.setGatewayEventSubstitutionListener(
          (GatewayEventSubstitutionFilter) getClass(substitutionFilterClass).newInstance());
    }
    // set dispatcher threads
    factory.setDispatcherThreads(numDispatcherThreadsForTheRun);
    AsyncEventQueue asyncChannel = factory.create(asyncChannelId,
        (AsyncEventListener) getClass(asyncListenerClass).newInstance());
  }

  private static Class getClass(String simpleClassName) throws Exception {
    String packagePrefix = "org.apache.geode.internal.cache.wan.";
    String className = packagePrefix + simpleClassName;
    Class clazz = null;
    clazz = Class.forName(className);
    return clazz;
  }

  public static void createAsyncEventQueueWithCustomListener(String asyncChannelId,
      boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation,
      boolean isPersistent, String diskStoreName, boolean isDiskSynchronous) {
    createAsyncEventQueueWithCustomListener(asyncChannelId, isParallel, maxMemory, batchSize,
        isConflation, isPersistent, diskStoreName, isDiskSynchronous,
        GatewaySender.DEFAULT_DISPATCHER_THREADS);
  }

  public static void createAsyncEventQueueWithCustomListener(String asyncChannelId,
      boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation,
      boolean isPersistent, String diskStoreName, boolean isDiskSynchronous, int nDispatchers) {

    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());

    try {
      createDiskStore(asyncChannelId, diskStoreName);

      AsyncEventListener asyncEventListener = new CustomAsyncEventListener();

      AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory,
          batchSize, isPersistent, diskStoreName);
      factory.setDispatcherThreads(nDispatchers);
      AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener);
    } finally {
      exp.remove();
    }
  }

  private static AsyncEventQueueFactory getInitialAsyncEventQueueFactory(boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isPersistent, String diskStoreName) {
    AsyncEventQueueFactory factory = cache.createAsyncEventQueueFactory();
    factory.setBatchSize(batchSize);
    factory.setPersistent(isPersistent);
    factory.setDiskStoreName(diskStoreName);
    factory.setMaximumQueueMemory(maxMemory);
    factory.setParallel(isParallel);
    return factory;
  }

  public static void createConcurrentAsyncEventQueue(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent,
      String diskStoreName, boolean isDiskSynchronous, int dispatcherThreads, OrderPolicy policy) {

    createDiskStore(asyncChannelId, diskStoreName);

    AsyncEventListener asyncEventListener = new MyAsyncEventListener();

    AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory,
        batchSize, isPersistent, diskStoreName);
    factory.setDiskSynchronous(isDiskSynchronous);
    factory.setBatchConflationEnabled(isConflation);
    factory.setOrderPolicy(policy);
    AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener);
  }

  public static String createAsyncEventQueueWithDiskStore(String asyncChannelId, boolean isParallel,
      Integer maxMemory, Integer batchSize, boolean isPersistent, String diskStoreName) {

    AsyncEventListener asyncEventListener = new MyAsyncEventListener();

    File persistentDirectory = null;
    if (diskStoreName == null) {
      persistentDirectory = new File(
          asyncChannelId + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum());
    } else {
      persistentDirectory = new File(diskStoreName);
    }
    LogWriterUtils.getLogWriter().info("The ds is : " + persistentDirectory.getName());
    persistentDirectory.mkdir();
    DiskStoreFactory dsf = cache.createDiskStoreFactory();
    File[] dirs1 = new File[] {persistentDirectory};

    AsyncEventQueueFactory factory = cache.createAsyncEventQueueFactory();
    factory.setBatchSize(batchSize);
    factory.setParallel(isParallel);
    if (isPersistent) {
      factory.setPersistent(isPersistent);
      factory.setDiskStoreName(dsf.setDiskDirs(dirs1).create(asyncChannelId).getName());
    }
    factory.setMaximumQueueMemory(maxMemory);
    // set dispatcher threads
    factory.setDispatcherThreads(numDispatcherThreadsForTheRun);
    AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener);
    return persistentDirectory.getName();
  }

  public static void pauseAsyncEventQueue(String asyncChannelId) {
    AsyncEventQueue theChannel = null;

    Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues();
    for (AsyncEventQueue asyncChannel : asyncEventChannels) {
      if (asyncChannelId.equals(asyncChannel.getId())) {
        theChannel = asyncChannel;
      }
    }

    ((AsyncEventQueueImpl) theChannel).getSender().pause();
  }

  public static void pauseAsyncEventQueueAndWaitForDispatcherToPause(String asyncChannelId) {
    AsyncEventQueue theChannel = null;

    Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues();
    for (AsyncEventQueue asyncChannel : asyncEventChannels) {
      if (asyncChannelId.equals(asyncChannel.getId())) {
        theChannel = asyncChannel;
        break;
      }
    }

    ((AsyncEventQueueImpl) theChannel).getSender().pause();

    ((AsyncEventQueueImpl) theChannel).getSender().getEventProcessor()
        .waitForDispatcherToPause();
  }

  public static void resumeAsyncEventQueue(String asyncQueueId) {
    AsyncEventQueue theQueue = null;

    Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues();
    for (AsyncEventQueue asyncChannel : asyncEventChannels) {
      if (asyncQueueId.equals(asyncChannel.getId())) {
        theQueue = asyncChannel;
      }
    }

    ((AsyncEventQueueImpl) theQueue).getSender().resume();
  }

  public static void waitForAsyncEventQueueSize(String senderId, int numQueueEntries,
      boolean localSize) {
    await()
        .untilAsserted(() -> checkAsyncEventQueueSize(senderId, numQueueEntries, localSize));
  }

  public static void checkAsyncEventQueueSize(String asyncQueueId, int numQueueEntries) {
    checkAsyncEventQueueSize(asyncQueueId, numQueueEntries, false);
  }

  public static void checkAsyncEventQueueSize(String asyncQueueId, int numQueueEntries,
      boolean localSize) {
    AsyncEventQueueImpl aeq = (AsyncEventQueueImpl) cache.getAsyncEventQueue(asyncQueueId);
    GatewaySender sender = aeq.getSender();

    if (sender.isParallel()) {
      Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues();
      Region queueRegion = queues.toArray(new RegionQueue[queues.size()])[0].getRegion();
      if (localSize) {
        queueRegion = PartitionRegionHelper.getLocalData(queueRegion);
      }
      assertEquals(numQueueEntries, queueRegion.size());
    } else {
      Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues();
      int size = 0;
      for (RegionQueue q : queues) {
        size += q.size();
      }
      assertEquals(numQueueEntries, size);
    }
  }

  public static void createPartitionedRegion(String regionName, String senderIds,
      Integer redundantCopies, Integer totalNumBuckets) {
    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        IgnoredException.addIgnoredException(PartitionOfflineException.class.getName());
    try {
      AttributesFactory fact = new AttributesFactory();
      if (senderIds != null) {
        StringTokenizer tokenizer = new StringTokenizer(senderIds, ",");
        while (tokenizer.hasMoreTokens()) {
          String senderId = tokenizer.nextToken();
          // GatewaySender sender = cache.getGatewaySender(senderId);
          // assertNotNull(sender);
          fact.addGatewaySenderId(senderId);
        }
      }
      PartitionAttributesFactory pfact = new PartitionAttributesFactory();
      pfact.setTotalNumBuckets(totalNumBuckets);
      pfact.setRedundantCopies(redundantCopies);
      pfact.setRecoveryDelay(0);
      fact.setPartitionAttributes(pfact.create());
      Region r = cache.createRegionFactory(fact.create()).create(regionName);
      assertNotNull(r);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void createPartitionedRegionWithAsyncEventQueue(String regionName,
      String asyncEventQueueId, Boolean offHeap) {
    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        IgnoredException.addIgnoredException(PartitionOfflineException.class.getName());
    try {
      AttributesFactory fact = new AttributesFactory();

      PartitionAttributesFactory pfact = new PartitionAttributesFactory();
      pfact.setTotalNumBuckets(16);
      fact.setPartitionAttributes(pfact.create());
      fact.setOffHeap(offHeap);
      Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId)
          .create(regionName);
      assertNotNull(r);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void createFixedPartitionedRegionWithAsyncEventQueue(String regionName,
      String asyncEventQueueId, String partitionName, final List<String> allPartitions,
      boolean offHeap) {
    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        IgnoredException.addIgnoredException(PartitionOfflineException.class.getName());
    try {
      AttributesFactory fact = new AttributesFactory();

      PartitionAttributesFactory pfact = new PartitionAttributesFactory();
      pfact.setTotalNumBuckets(16);
      pfact.addFixedPartitionAttributes(
          FixedPartitionAttributes.createFixedPartition(partitionName, true));
      pfact.setPartitionResolver(new MyFixedPartitionResolver(allPartitions));
      fact.setPartitionAttributes(pfact.create());
      fact.setOffHeap(offHeap);
      Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId)
          .create(regionName);
      assertNotNull(r);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void createColocatedPartitionedRegionWithAsyncEventQueue(String regionName,
      String asyncEventQueueId, Integer totalNumBuckets, String colocatedWith) {

    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());
    IgnoredException exp1 =
        IgnoredException.addIgnoredException(PartitionOfflineException.class.getName());
    try {
      AttributesFactory fact = new AttributesFactory();

      PartitionAttributesFactory pfact = new PartitionAttributesFactory();
      pfact.setTotalNumBuckets(totalNumBuckets);
      pfact.setColocatedWith(colocatedWith);
      fact.setPartitionAttributes(pfact.create());
      Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId)
          .create(regionName);
      assertNotNull(r);
    } finally {
      exp.remove();
      exp1.remove();
    }
  }

  public static void createPartitionedRegionWithCacheLoaderAndAsyncQueue(String regionName,
      String asyncEventQueueId) {

    AttributesFactory fact = new AttributesFactory();

    PartitionAttributesFactory pfact = new PartitionAttributesFactory();
    pfact.setTotalNumBuckets(16);
    fact.setPartitionAttributes(pfact.create());
    // set the CacheLoader implementation
    fact.setCacheLoader(new MyCacheLoader());
    Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId)
        .create(regionName);
    assertNotNull(r);
  }

  /**
   * Create PartitionedRegion with 1 redundant copy
   */
  public static void createPRWithRedundantCopyWithAsyncEventQueue(String regionName,
      String asyncEventQueueId, Boolean offHeap) throws InterruptedException {
    IgnoredException exp =
        IgnoredException.addIgnoredException(ForceReattemptException.class.getName());


    CountDownLatch recoveryDone = new CountDownLatch(2);

    ResourceObserver observer = new InternalResourceManager.ResourceObserverAdapter() {
      @Override
      public void recoveryFinished(Region region) {
        recoveryDone.countDown();
      }
    };
    InternalResourceManager.setResourceObserver(observer);


    try {
      AttributesFactory fact = new AttributesFactory();

      PartitionAttributesFactory pfact = new PartitionAttributesFactory();
      pfact.setTotalNumBuckets(16);
      pfact.setRedundantCopies(1);
      fact.setPartitionAttributes(pfact.create());
      fact.setOffHeap(offHeap);
      Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId)
          .create(regionName);
      assertNotNull(r);
      recoveryDone.await();
    } finally {
      exp.remove();
    }
  }

  public static void createPartitionedRegionAccessorWithAsyncEventQueue(String regionName,
      String asyncEventQueueId) {
    AttributesFactory fact = new AttributesFactory();
    PartitionAttributesFactory pfact = new PartitionAttributesFactory();
    pfact.setTotalNumBuckets(16);
    pfact.setLocalMaxMemory(0);
    fact.setPartitionAttributes(pfact.create());
    Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId)
        .create(regionName);
    // fact.create()).create(regionName);
    assertNotNull(r);
  }

  protected static void createCache(Integer locPort) {
    AsyncEventQueueTestBase test = new AsyncEventQueueTestBase();
    Properties props = test.getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "localhost[" + locPort + "]");
    InternalDistributedSystem ds = test.getSystem(props);
    cache = CacheFactory.create(ds);
  }

  public static void createCacheWithoutLocator(Integer mCastPort) {
    AsyncEventQueueTestBase test = new AsyncEventQueueTestBase();
    Properties props = test.getDistributedSystemProperties();
    props.setProperty(MCAST_PORT, "" + mCastPort);
    InternalDistributedSystem ds = test.getSystem(props);
    cache = CacheFactory.create(ds);
  }

  public static void checkAsyncEventQueueStats(String queueId, final int queueSize,
      int secondaryQueueSize, final int eventsReceived, final int eventsQueued,
      final int eventsDistributed) {
    Set<AsyncEventQueue> asyncQueues = cache.getAsyncEventQueues();
    AsyncEventQueue queue = null;
    boolean isParallel = false;
    for (AsyncEventQueue q : asyncQueues) {
      isParallel = q.isParallel();
      if (q.getId().equals(queueId)) {
        queue = q;
        break;
      }
    }
    final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics();
    await()
        .untilAsserted(
            () -> assertEquals("Expected queue entries: " + queueSize + " but actual entries: "
                + statistics.getEventQueueSize(), queueSize, statistics.getEventQueueSize()));
    if (isParallel) {
      await().untilAsserted(() -> {
        assertEquals(
            "Expected events in the secondary queue is " + secondaryQueueSize + ", but actual is "
                + statistics.getSecondaryEventQueueSize(),
            secondaryQueueSize, statistics.getSecondaryEventQueueSize());
      });
    } else {
      // for serial queue, evenvSecondaryQueueSize is not used
      assertEquals(0, statistics.getSecondaryEventQueueSize());
    }
    assertEquals(queueSize, statistics.getEventQueueSize());
    assertEquals(eventsReceived, statistics.getEventsReceived());
    assertEquals(eventsQueued, statistics.getEventsQueued());
    assert (statistics.getEventsDistributed() >= eventsDistributed);
  }

  public static void checkAsyncEventQueueConflatedStats(String asyncEventQueueId,
      final int eventsConflated) {
    Set<AsyncEventQueue> queues = cache.getAsyncEventQueues();
    AsyncEventQueue queue = null;
    for (AsyncEventQueue q : queues) {
      if (q.getId().equals(asyncEventQueueId)) {
        queue = q;
        break;
      }
    }
    final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics();
    assertEquals(eventsConflated, statistics.getEventsNotQueuedConflated());
  }

  public static void checkAsyncEventQueueStats_Failover(String asyncEventQueueId,
      final int eventsReceived) {
    Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues();
    AsyncEventQueue queue = null;
    for (AsyncEventQueue q : asyncEventQueues) {
      if (q.getId().equals(asyncEventQueueId)) {
        queue = q;
        break;
      }
    }
    final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics();

    assertEquals(eventsReceived, statistics.getEventsReceived());
    assertEquals(eventsReceived,
        (statistics.getEventsQueued() + statistics.getUnprocessedTokensAddedByPrimary()
            + statistics.getUnprocessedEventsRemovedByPrimary()));
  }

  public static void checkAsyncEventQueueBatchStats(String asyncQueueId, final int batches) {
