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

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache.wan.GatewaySender.DEFAULT_BATCH_SIZE;
import static org.apache.geode.cache.wan.GatewaySender.GET_TRANSACTION_EVENTS_FROM_QUEUE_WAIT_TIME_MS;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;

import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.AbstractBucketRegionQueue;
import org.apache.geode.internal.cache.BucketNotFoundException;
import org.apache.geode.internal.cache.BucketRegion;
import org.apache.geode.internal.cache.BucketRegionQueue;
import org.apache.geode.internal.cache.ColocationHelper;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.DiskRegionStats;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.InternalRegionFactory;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.internal.cache.PrimaryBucketException;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
import org.apache.geode.internal.cache.wan.GatewaySenderException;
import org.apache.geode.internal.cache.wan.GatewaySenderStats;
import org.apache.geode.internal.size.SingleObjectSizer;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.StoppableCondition;
import org.apache.geode.internal.util.concurrent.StoppableReentrantLock;
import org.apache.geode.logging.internal.executors.LoggingExecutors;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.internal.beans.AsyncEventQueueMBean;
import org.apache.geode.management.internal.beans.GatewaySenderMBean;

public class ParallelGatewaySenderQueue implements RegionQueue {
  protected static final Logger logger = LogService.getLogger();
  protected final Map<String, PartitionedRegion> userRegionNameToShadowPRMap =
      new ConcurrentHashMap<>();
  private static final String SHADOW_BUCKET_PATH_PREFIX =
      SEPARATOR + PartitionedRegionHelper.PR_ROOT_REGION_NAME + SEPARATOR;

  // <PartitionedRegion, Map<Integer, List<Object>>>
  private final Map<String, Map<Integer, List<Object>>> regionToDispatchedKeysMap =
      new ConcurrentHashMap<String, Map<Integer, List<Object>>>();

  protected final StoppableReentrantLock buckToDispatchLock;
  private final StoppableCondition regionToDispatchedKeysMapEmpty;
  protected final StoppableReentrantLock queueEmptyLock;
  private volatile boolean isQueueEmpty = true;

  private final boolean cleanQueues;

  @VisibleForTesting
  boolean getCleanQueues() {
    return cleanQueues;
  }


  private final boolean asyncEvent;

  /**
   * False signal is fine on this condition. As processor will loop again and find out if it was a
   * false signal. However, make sure that whatever scenario can cause an entry to be peeked shoudld
   * signal the processor to unblock.
   */
  private final StoppableCondition queueEmptyCondition;

  protected final GatewaySenderStats stats;

  protected volatile boolean resetLastPeeked = false;

  /**
   * There will be one shadow pr for each of the the PartitionedRegion which has added the
   * GatewaySender Fix for Bug#45917 We maintain a tempQueue to queue events when buckets are not
   * available locally.
   */
  private final ConcurrentMap<Integer, BlockingQueue<GatewaySenderEventImpl>> bucketToTempQueueMap =
      new ConcurrentHashMap<>();

  /**
   * The default frequency (in milliseconds) at which a message will be sent by the primary to all
   * the secondary nodes to remove the events which have already been dispatched from the queue.
   */
  public static final int DEFAULT_MESSAGE_SYNC_INTERVAL = 10;

  // TODO:REF: how to change the message sync interval ? should it be common for serial and parallel
  @MutableForTesting
  protected static volatile int messageSyncInterval = DEFAULT_MESSAGE_SYNC_INTERVAL;

  // TODO:REF: name change for thread, as it appears in the log
  private BatchRemovalThread removalThread = null;

  protected BlockingQueue<GatewaySenderEventImpl> peekedEvents =
      new LinkedBlockingQueue<>();

  /**
   * The peekedEventsProcessing queue is used when the batch size is reduced due to a
   * MessageTooLargeException
   */
  private final BlockingQueue<GatewaySenderEventImpl> peekedEventsProcessing =
      new LinkedBlockingQueue<>();

  /**
   * The peekedEventsProcessingInProgress boolean denotes that processing existing peeked events is
   * in progress
   */
  private boolean peekedEventsProcessingInProgress = false;

  public final AbstractGatewaySender sender;

  public static final int WAIT_CYCLE_SHADOW_BUCKET_LOAD = 10;

  public static final String QSTRING = "_PARALLEL_GATEWAY_SENDER_QUEUE";

  /**
   * Fixed size Thread pool for conflating the events in the queue. The size of the thread pool is
   * set to the number of processors available to the JVM. There will be one thread pool per
   * ParallelGatewaySender on a node.
   */
  private ExecutorService conflationExecutor;

  /**
   * This class carries out the actual removal of the previousTailKey from QPR. The class implements
   * Runnable and the destroy operation is done in the run method. The Runnable is executed by the
   * one of the threads in the conflation thread pool configured above.
   */
  private class ConflationHandler implements Runnable {
    Conflatable conflatableObject;

    Long previousTailKeyTobeRemoved;

    int bucketId;

    public ConflationHandler(Conflatable conflatableObject, int bId, Long previousTailKey) {
      this.conflatableObject = conflatableObject;
      previousTailKeyTobeRemoved = previousTailKey;
      bucketId = bId;
    }

    @Override
    public void run() {
      GatewaySenderEventImpl event = (GatewaySenderEventImpl) conflatableObject;
      String regionPath =
          ColocationHelper.getLeaderRegion((PartitionedRegion) event.getRegion()).getFullPath();
      PartitionedRegion prQ = userRegionNameToShadowPRMap.get(regionPath);
      try {
        destroyEventFromQueue(prQ, bucketId, previousTailKeyTobeRemoved);
      } catch (EntryNotFoundException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("{}: Not conflating {} due to EntryNotFoundException", this,
              conflatableObject.getKeyToConflate());
        }
      }
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Conflated {} for key={} in queue for region={}", this,
            conflatableObject.getValueToConflate(), conflatableObject.getKeyToConflate(),
            prQ.getName());
      }
    }

    private Object deserialize(Object serializedBytes) {
      Object deserializedObject = serializedBytes;
      if (serializedBytes instanceof byte[]) {
        byte[] serializedBytesCast = (byte[]) serializedBytes;
        // This is a debugging method so ignore all exceptions like
        // ClassNotFoundException
        try {
          deserializedObject = EntryEventImpl.deserialize(serializedBytesCast);
        } catch (Exception ignore) {
        }
      }
      return deserializedObject;
    }
  }

  protected final int index;

  protected final int nDispatcher;

  private final MetaRegionFactory metaRegionFactory;

  public ParallelGatewaySenderQueue(AbstractGatewaySender sender, Set<Region<?, ?>> userRegions,
      int idx,
      int nDispatcher, boolean cleanQueues) {
    this(sender, userRegions, idx, nDispatcher, new MetaRegionFactory(), cleanQueues);
  }

  ParallelGatewaySenderQueue(AbstractGatewaySender sender, Set<Region<?, ?>> userRegions, int idx,
      int nDispatcher, MetaRegionFactory metaRegionFactory, boolean cleanQueues) {

    this.metaRegionFactory = metaRegionFactory;

    this.cleanQueues = cleanQueues;

    index = idx;
    this.nDispatcher = nDispatcher;
    stats = sender.getStatistics();
    this.sender = sender;

    asyncEvent = this.sender.getId().contains(AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX);

    List<Region> listOfRegions = new ArrayList<>(userRegions);
    Collections.sort(listOfRegions, (o1, o2) -> o1.getFullPath().compareTo(o2.getFullPath()));

    for (Region userRegion : listOfRegions) {
      if (userRegion instanceof PartitionedRegion) {
        addShadowPartitionedRegionForUserPR((PartitionedRegion) userRegion);
        if (index == 0 && getRegion(userRegion.getFullPath()) != null) {
          stats.incQueueSize(getRegion(userRegion.getFullPath()).getLocalSize());
        }
      } else {
        // Fix for Bug#51491. Once decided to support this configuration we have call
        // addShadowPartitionedRegionForUserRR

        if (asyncEvent) {
          throw new AsyncEventQueueConfigurationException(String.format(
              "Parallel Async Event Queue %s can not be used with replicated region %s",
              AsyncEventQueueImpl.getAsyncEventQueueIdFromSenderId(this.sender.getId()),
              userRegion.getFullPath()));
        }
        throw new GatewaySenderConfigurationException(
            String.format("Parallel Gateway Sender %s can not be used with replicated region %s",
                this.sender.getId(), userRegion.getFullPath()));
      }
    }

    buckToDispatchLock = new StoppableReentrantLock(sender.getCancelCriterion());
    regionToDispatchedKeysMapEmpty = buckToDispatchLock.newCondition();

    queueEmptyLock = new StoppableReentrantLock(sender.getCancelCriterion());
    queueEmptyCondition = queueEmptyLock.newCondition();

    // initialize the conflation thread pool if conflation is enabled
    if (sender.isBatchConflationEnabled()) {
      initializeConflationThreadPool();
    }
  }

  /** Start the background batch removal thread. */
  public void start() {
    // at present, this won't be accessed by multiple threads,
    // still, it is safer approach to synchronize it
    synchronized (ParallelGatewaySenderQueue.class) {
      if (removalThread == null) {
        removalThread = new BatchRemovalThread(sender.getCache(), this);
        removalThread.start();
      }
    }
  }

  public void addShadowPartitionedRegionForUserRR(DistributedRegion userRegion) {
    sender.getLifeCycleLock().writeLock().lock();
    PartitionedRegion prQ = null;

    if (logger.isDebugEnabled()) {
      logger.debug(
          "addShadowPartitionedRegionForUserRR: Going to create shadowpr for userRegion {}",
          userRegion.getFullPath());
    }

    try {
      String regionName = userRegion.getFullPath();

      if (userRegionNameToShadowPRMap.containsKey(regionName)) {
        return;
      }

      InternalCache cache = sender.getCache();
      final String prQName = getQueueName(sender.getId(), userRegion.getFullPath());
      prQ = (PartitionedRegion) cache.getRegion(prQName);
      if (prQ == null) {
        InternalRegionFactory fact = cache.createInternalRegionFactory();
        // Fix for 48621 - don't enable concurrency checks
        // for queue buckets., event with persistence
        fact.setConcurrencyChecksEnabled(false);
        PartitionAttributesFactory pfact = new PartitionAttributesFactory();
        pfact.setTotalNumBuckets(sender.getMaxParallelismForReplicatedRegion());
        int localMaxMemory =
            userRegion.getDataPolicy().withStorage() ? sender.getMaximumQueueMemory() : 0;
        pfact.setLocalMaxMemory(localMaxMemory);
        pfact.setRedundantCopies(3); // TODO:Kishor : THis need to be handled nicely
        pfact.setPartitionResolver(new RREventIDResolver());
        if (sender.isPersistenceEnabled()) {
          fact.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
        }

        fact.setDiskStoreName(sender.getDiskStoreName());

        // if persistence is enabled, set the diskSyncronous to whatever user
        // has set
        // else set it to false
        // optimize with above check of enable persistence
        if (sender.isPersistenceEnabled()) {
          fact.setDiskSynchronous(sender.isDiskSynchronous());
        } else {
          fact.setDiskSynchronous(false);
        }

        // allow for no overflow directory
        EvictionAttributes ea = EvictionAttributes.createLIFOMemoryAttributes(
            sender.getMaximumQueueMemory(), EvictionAction.OVERFLOW_TO_DISK);

        fact.setEvictionAttributes(ea);
        fact.setPartitionAttributes(pfact.create());

        final RegionAttributes ra = fact.getCreateAttributes();

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Attempting to create queue region: {}", this, prQName);
        }

        ParallelGatewaySenderQueueMetaRegion meta =
            new ParallelGatewaySenderQueueMetaRegion(prQName, ra, null, cache, sender,
                sender.getStatisticsClock());

        fact.setInternalMetaRegion(meta).setDestroyLockFlag(true)
            .setSnapshotInputStream(null).setImageTarget(null);
        prQ = (PartitionedRegion) fact.create(prQName);

        if (logger.isDebugEnabled()) {
          logger.debug("Region created  : {} partition Attributes : {}", prQ,
              prQ.getPartitionAttributes());
        }

        // TODO This should not be set on the PR but on the GatewaySender
        prQ.enableConflation(sender.isBatchConflationEnabled());

        // Before going ahead, make sure all the buckets of shadowPR are
        // loaded
        // and primary nodes have been decided.
        // This is required in case of persistent PR and sender.
        if (prQ.getLocalMaxMemory() != 0) {
          for (final Integer integer : prQ.getRegionAdvisor().getBucketSet()) {
          }
        }
        // In case of Replicated Region it may not be necessary.

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Created queue region: {}", this, prQ);
        }
      } else {
        // in case shadowPR exists already (can be possible when sender is
        // started from stop operation)
        // for first processor only
        if (index == 0) {
          handleShadowPRExistsScenario(cache, prQ);
        }
      }
      /*
       * Here, enqueueTempEvents need to be invoked when a sender is already running and userPR is
       * created later. When the flow comes here through start() method of sender i.e. userPR
       * already exists and sender is started later, the enqueueTempEvents is done in the start()
       * method of ParallelGatewaySender
       */
      if ((index == nDispatcher - 1) && sender.isRunning()) {
        sender.enqueueTempEvents();
      }
    } finally {
      if (prQ != null) {
        userRegionNameToShadowPRMap.put(userRegion.getFullPath(), prQ);
      }
      sender.getLifeCycleLock().writeLock().unlock();
    }
  }

  private static String convertPathToName(String fullPath) {
    return "";
  }

  public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR) {
    addShadowPartitionedRegionForUserPR(userPR, null);
  }

  public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
      PartitionedRegion childPR) {
    if (logger.isDebugEnabled()) {
      logger.debug(
          "{} addShadowPartitionedRegionForUserPR: Attempting to create queue region: {}; child region: {}",
          this, userPR.getDisplayName(), childPR == null ? "null" : childPR.getDisplayName());
    }
    sender.getLifeCycleLock().writeLock().lock();

    PartitionedRegion prQ = null;
    try {
      String regionName = userPR.getFullPath();
      // Find if there is any parent region for this userPR
      // if there is then no need to add another q for the same
      String leaderRegionName = ColocationHelper.getLeaderRegion(userPR).getFullPath();
      if (!regionName.equals(leaderRegionName)) {
        // Fix for defect #50364. Allow user to attach GatewaySender to child PR (without attaching
        // to leader PR)
        // though, internally, colocate the GatewaySender's shadowPR with the leader PR in
        // colocation chain
        if (!userRegionNameToShadowPRMap.containsKey(leaderRegionName)) {
          addShadowPartitionedRegionForUserPR(ColocationHelper.getLeaderRegion(userPR), userPR);
        }
        return;
      }

      if (userRegionNameToShadowPRMap.containsKey(regionName)) {
        return;
      }

      if ((childPR == null ? userPR : childPR).getDataPolicy().withPersistence()
          && !sender.isPersistenceEnabled()) {
        throw new GatewaySenderException(
            String.format(
                "Non persistent gateway sender %s can not be attached to persistent region %s",
                sender.getId(), userPR.getFullPath()));
      }

      InternalCache cache = sender.getCache();
      boolean isAccessor = (userPR.getLocalMaxMemory() == 0);

      final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());

      prQ = (PartitionedRegion) cache.getRegion(prQName, true);
      if (prQ != null && prQ.isDestroyed()) {
        PartitionedRegion oldPrQ = prQ;
        while (oldPrQ == prQ) {
          try {
            Thread.sleep(50);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
          }
          if (logger.isDebugEnabled()) {
            logger.debug("wait for destroy to finish");
          }
          prQ = (PartitionedRegion) cache.getRegion(prQName, true);
        }
      }

      if ((prQ != null) && (index == 0) && cleanQueues) {
        cleanOverflowStats(cache);
        prQ.clearPREvictionControllerFromDiskInitialization();
        prQ.destroyRegion(null);
        prQ = null;
      }

      if (prQ == null) {
        RegionShortcut regionShortcut;
        if (sender.isPersistenceEnabled() && !isAccessor) {
          regionShortcut = RegionShortcut.PARTITION_PERSISTENT;
        } else {
          regionShortcut = RegionShortcut.PARTITION;
        }
        InternalRegionFactory fact = cache.createInternalRegionFactory(regionShortcut);

        fact.setConcurrencyChecksEnabled(false);
        PartitionAttributesFactory pfact = new PartitionAttributesFactory();
        pfact.setTotalNumBuckets(userPR.getTotalNumberOfBuckets());
        pfact.setRedundantCopies(userPR.getRedundantCopies());
        pfact.setColocatedWith(regionName);
        // EITHER set localMaxMemory to 0 for accessor node
        // OR override shadowPRs default local max memory with the sender's max
        // queue memory (Fix for bug#44254)
        int localMaxMemory = isAccessor ? 0 : sender.getMaximumQueueMemory();
        pfact.setLocalMaxMemory(localMaxMemory);

        pfact.setStartupRecoveryDelay(userPR.getPartitionAttributes().getStartupRecoveryDelay());
        pfact.setRecoveryDelay(userPR.getPartitionAttributes().getRecoveryDelay());

        fact.setDiskStoreName(sender.getDiskStoreName());

        // if persistence is enabled, set the diskSyncronous to whatever user has set
        // else set it to false
        if (sender.isPersistenceEnabled()) {
          fact.setDiskSynchronous(sender.isDiskSynchronous());
        } else {
          fact.setDiskSynchronous(false);
        }

        // allow for no overflow directory
        EvictionAttributes ea = EvictionAttributes.createLIFOMemoryAttributes(
            sender.getMaximumQueueMemory(), EvictionAction.OVERFLOW_TO_DISK);

        fact.setEvictionAttributes(ea);
        fact.setPartitionAttributes(pfact.create());

        final RegionAttributes ra = fact.getCreateAttributes();

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Attempting to create queue region: {}", this, prQName);
        }

        ParallelGatewaySenderQueueMetaRegion meta =
            metaRegionFactory.newMetataRegion(cache, prQName, ra, sender);

        fact.setInternalMetaRegion(meta);
        fact.setDestroyLockFlag(true);
        fact.setInternalRegion(true);
        fact.setSnapshotInputStream(null);
        fact.setImageTarget(null);
        prQ = (PartitionedRegion) fact.create(prQName);
        // at this point we should be able to assert prQ == meta;

        if (prQ == null) {
          return;
        }
        // TODO This should not be set on the PR but on the GatewaySender
        prQ.enableConflation(sender.isBatchConflationEnabled());
        if (isAccessor) {
          return; // return from here if accessor node
        }

        // Add the overflow statistics to the mbean
        addOverflowStatisticsToMBean(cache, prQ);

        // Wait for buckets to be recovered.
        prQ.shadowPRWaitForBucketRecovery();

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Created queue region: {}", this, prQ);
        }

      } else {
        if (isAccessor) {
          return; // return from here if accessor node
        }
        // in case shadowPR exists already (can be possible when sender is
        // started from stop operation)
        // for first parallelGatewaySenderQueue only
        if (index == 0) {
          handleShadowPRExistsScenario(cache, prQ);
        }

      }

    } finally {
      if (prQ != null) {
        userRegionNameToShadowPRMap.put(userPR.getFullPath(), prQ);
        prQ.setSentGatewaySenderStoppedMessage(false);
        if (prQ.getDataStore() != null) {
          final Set<BucketRegion> buckets = prQ.getDataStore().getAllLocalBucketRegions();
          for (BucketRegion br : buckets) {
            br.setReceivedGatewaySenderStoppedMessage(false);
          }
        }
      }
      /*
       * Here, enqueueTempEvents need to be invoked when a sender is already running and userPR is
       * created later. When the flow comes here through start() method of sender i.e. userPR
       * already exists and sender is started later, the enqueueTempEvents is done in the start()
       * method of ParallelGatewaySender
       */
      if ((index == nDispatcher - 1) && sender.isRunning()) {
        sender.enqueueTempEvents();
      }
      afterRegionAdd(userPR);
      sender.getLifeCycleLock().writeLock().unlock();
    }
  }

  private void addOverflowStatisticsToMBean(Cache cache, PartitionedRegion prQ) {
    // Get the appropriate mbean and add the eviction and disk region stats to it
    ManagementService service = ManagementService.getManagementService(cache);
    if (asyncEvent) {
      AsyncEventQueueMBean bean = (AsyncEventQueueMBean) service.getLocalAsyncEventQueueMXBean(
          AsyncEventQueueImpl.getAsyncEventQueueIdFromSenderId(sender.getId()));

      if (bean != null) {
        // Add the eviction stats
        bean.getBridge().addOverflowStatistics(prQ.getEvictionStatistics());

        // Add the disk region stats
        bean.getBridge().addOverflowStatistics(prQ.getDiskRegionStats().getStats());
      }
    } else {
      GatewaySenderMBean bean =
          (GatewaySenderMBean) service.getLocalGatewaySenderMXBean(sender.getId());

      if (bean != null) {
        // Add the eviction stats
        bean.getBridge().addOverflowStatistics(prQ.getEvictionStatistics());

        // Add the disk region stats
        bean.getBridge().addOverflowStatistics(prQ.getDiskRegionStats().getStats());
      }
    }
  }

  private void cleanOverflowStats(Cache cache) {
    ManagementService service = ManagementService.getManagementService(cache);
    if (!asyncEvent) {
      GatewaySenderMBean bean =
          (GatewaySenderMBean) service.getLocalGatewaySenderMXBean(sender.getId());
      if (bean != null) {
        bean.getBridge().clearOverflowStatistics();
      }
    }

  }


  /**
   * This will be case when the sender is started again after stop operation.
   */
  private void handleShadowPRExistsScenario(Cache cache, PartitionedRegion prQ) {
    // Note: The region will not be null if the sender is started again after stop operation
    if (logger.isDebugEnabled()) {
      logger.debug("{}: No need to create the region as the region has been retrieved: {}", this,
          prQ);
    }
  }

  protected void afterRegionAdd(PartitionedRegion userPR) {
    // nothing
  }

  /**
   * Initialize the thread pool, setting the number of threads that is equal to the number of
   * processors available to the JVM.
   */
  private void initializeConflationThreadPool() {
    int poolSize = Runtime.getRuntime().availableProcessors();
    conflationExecutor =
        LoggingExecutors.newFixedThreadPool(poolSize, "WAN Queue Conflation Thread", true);
  }

  /**
   * Cleans up the conflation thread pool. Initially, shutdown is done to avoid accepting any newly
   * submitted tasks. Wait a while for existing tasks to terminate. If the existing tasks still
   * don't complete, cancel them by calling shutdownNow.
   */
  private void cleanupConflationThreadPool(AbstractGatewaySender sender) {
    if (conflationExecutor == null) {
      return;
    }
    conflationExecutor.shutdown();// Disable new tasks from being submitted

    try {
      if (!conflationExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
        conflationExecutor.shutdownNow(); // Cancel currently executing tasks
        // Wait a while for tasks to respond to being cancelled
        if (!conflationExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
          logger.warn("Conflation thread pool did not terminate for the GatewaySender : {}",
              (sender == null ? "all" : sender));
        }
      }
    } catch (InterruptedException e) {
      // (Re-)Cancel if current thread also interrupted
      conflationExecutor.shutdownNow();
      // Preserve interrupt status
      Thread.currentThread().interrupt();
    }
  }

  @Override
  public boolean put(Object object) throws InterruptedException, CacheException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    boolean putDone = false;
    // Can this region ever be null? Should we work with regionName and not with region
    // instance.
    // It can't be as put is happening on the region and its still under process
    GatewaySenderEventImpl value = (GatewaySenderEventImpl) object;

    boolean isDREvent = isDREvent(sender.getCache(), value);

    String regionPath = value.getRegionPath();
    if (!isDREvent) {
      Region region = sender.getCache().getRegion(regionPath, true);
      regionPath = ColocationHelper.getLeaderRegion((PartitionedRegion) region).getFullPath();
    }
    if (isDebugEnabled) {
      logger.debug("Put is for the region {}", regionPath);
    }
    if (!userRegionNameToShadowPRMap.containsKey(regionPath)) {
      if (isDebugEnabled) {
        logger.debug("The userRegionNameToshadowPRMap is {}", userRegionNameToShadowPRMap);
      }
      logger.warn(
          "GatewaySender: Not queuing the event {}, as the region for which this event originated is not yet configured in the GatewaySender",
          value);
      // does not put into queue
      return false;
    }

    PartitionedRegion prQ = userRegionNameToShadowPRMap.get(regionPath);
    int bucketId = value.getBucketId();
    Object key = null;
    if (!isDREvent) {
      key = value.getShadowKey();

      if ((Long) key == -1) {
        // In case of parallel we don't expect
        // the key to be not set. If it is the case then the event must be coming
        // through listener, so return.
        if (isDebugEnabled) {
          logger.debug("ParallelGatewaySenderOrderedQueue not putting key {} : Value : {}", key,
              value);
        }
        // does not put into queue
        return false;
      }
    } else {
      key = value.getEventId();
    }

    if (isDebugEnabled) {
      logger.debug("ParallelGatewaySenderOrderedQueue putting key {} : Value : {}", key, value);
    }
    AbstractBucketRegionQueue brq =
        (AbstractBucketRegionQueue) prQ.getDataStore().getLocalBucketById(bucketId);

    try {
      if (brq == null) {
        // Full path of the bucket
        final String bucketFullPath = SHADOW_BUCKET_PATH_PREFIX + prQ.getBucketName(bucketId);

        // Set the threadInitLevel to BEFORE_INITIAL_IMAGE.
        final InitializationLevel oldLevel =
            LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);

        try {
          brq = (AbstractBucketRegionQueue) prQ.getCache().getInternalRegionByPath(bucketFullPath);
          if (isDebugEnabled) {
            logger.debug(
                "ParallelGatewaySenderOrderedQueue : The bucket in the cache is bucketRegionName : {} bucket : {}",
                bucketFullPath, brq);
          }
          if (brq != null) {
            boolean intializingLocked = brq.lockWhenRegionIsInitializing();
            brq.getInitializationLock().readLock().lock();
            try {
              putIntoBucketRegionQueue(brq, key, value);
              putDone = true;
            } finally {
              brq.getInitializationLock().readLock().unlock();
              if (intializingLocked) {
                brq.unlockWhenRegionIsInitializing();
              }
            }
          } else if (isDREvent) {
            // in case of DR with PGS, if shadow bucket is not found event after
            // above search then it means that bucket is not intended for this
            // node. So lets not add this event in temp queue event as we are
            // doing it for PRevent
            // does not put onto the queue
          } else {
            // We have to handle the case where brq is null because the
            // colocation
