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

import static org.apache.geode.internal.statistics.StatisticsClockFactory.disabledClock;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
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.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;

import org.apache.geode.InternalGemFireError;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheRuntimeException;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.InterestRegistrationEvent;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.ResultSender;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.query.QueryInvalidException;
import org.apache.geode.cache.query.internal.QCompiler;
import org.apache.geode.cache.query.internal.index.IndexCreationData;
import org.apache.geode.cache.query.internal.index.PartitionedIndex;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.OperationExecutors;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.BucketRegion.RawValue;
import org.apache.geode.internal.cache.PartitionedRegion.BucketLock;
import org.apache.geode.internal.cache.PartitionedRegion.SizeEntry;
import org.apache.geode.internal.cache.backup.BackupService;
import org.apache.geode.internal.cache.execute.BucketMovedException;
import org.apache.geode.internal.cache.execute.PartitionedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
import org.apache.geode.internal.cache.execute.metrics.FunctionStats;
import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionFunctionStreamingMessage;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionObserver;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionObserverHolder;
import org.apache.geode.internal.cache.partitioned.RedundancyAlreadyMetException;
import org.apache.geode.internal.cache.partitioned.RemoveBucketMessage;
import org.apache.geode.internal.cache.partitioned.RemoveBucketMessage.RemoveBucketResponse;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock.StoppableReadLock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock.StoppableWriteLock;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implementation of DataStore (DS) for a PartitionedRegion (PR). This will be import
 * org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor; import
 * org.apache.geode.internal.cache.wan.GatewaySenderEventImpl; import
 * org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderImpl; import
 * org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue; accessed via accessor of
 * PartitionedRegion or PartionService thread which will handle remote calls to this DataStore from
 * other nodes participating in this PartitionedRegion.
 *
 */
public class PartitionedRegionDataStore implements HasCachePerfStats {
  private static final Logger logger = LogService.getLogger();

  /** PR reference for this DataStore */
  protected final PartitionedRegion partitionedRegion;

  /**
   * Total memory used by this partition. Only used for statistics.
   */
  private final AtomicLong bytesInUse = new AtomicLong(0);

  /**
   * CacheLoader of this PRDataStore
   */
  private CacheLoader loader;

  /**
   * Anyone creating a bucket must hold the read lock. Anyone deleting a bucket must hold the write
   * lock.
   */
  final StoppableReentrantReadWriteLock bucketCreationLock;

  /**
   * Maps each bucket to a real region that contains actual key/value entries for this PR instance
   * <p>
   * Keys are instances of {@link Integer}. Values are instances of (@link BucketRegion}.
   */
  private final ConcurrentMap<Integer, BucketRegion> localBucket2RegionMap;

  /**
   * A counter of the number of concurrent bucket creates in progress on this node
   */
  final AtomicInteger bucketCreatesInProgress = new AtomicInteger();

  /**
   * Variable used to determine when the data store is over the local max memory limit
   */
  private boolean exceededLocalMaxMemoryLimit = false;

  /**
   * Maximum number of bytes this dataStore has for storage
   */
  private final long maximumLocalBytes;

  private final CachePerfStats bucketStats;

  /**
   * The keysOfInterest contains a set of all keys in which any client has interest in this PR.
   */
  final ConcurrentMap keysOfInterest;

  private final Object keysOfInterestLock = new Object();

  private final StatisticsClock statisticsClock;

  /**
   * Update an entry's last access time if a client is interested in the entry.
   */
  private static final boolean UPDATE_ACCESS_TIME_ON_INTEREST =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "updateAccessTimeOnClientInterest");


  // Only for testing
  PartitionedRegionDataStore() {
    statisticsClock = disabledClock();
    bucketCreationLock = null;
    bucketStats = null;
    partitionedRegion = null;
    maximumLocalBytes = -1;
    localBucket2RegionMap = new ConcurrentHashMap<>();
    keysOfInterest = null;
  }

  /**
   * Creates PartitionedRegionDataStore for dataStorage of PR and starts a PartitionService to
   * handle remote operations on this DataStore from other participating nodes.
   *
   * @param pr PartitionedRegion associated with this DataStore.
   */
  PartitionedRegionDataStore(final PartitionedRegion pr, StatisticsClock statisticsClock) {
    this.statisticsClock = statisticsClock;
    final int bucketCount = pr.getTotalNumberOfBuckets();
    localBucket2RegionMap = new ConcurrentHashMap<>(bucketCount);
    partitionedRegion = pr;
    bucketCreationLock = new StoppableReentrantReadWriteLock(pr.getCancelCriterion());
    if (pr.getAttributes().getCacheLoader() != null) {
      loader = pr.getAttributes().getCacheLoader();
      if (logger.isDebugEnabled()) {
        logger.debug("Installing cache loader from partitioned region attributes: {}", loader);
      }
    }
    // this.maximumLocalBytes = (long) (pr.getLocalMaxMemory() *
    // PartitionedRegionHelper.BYTES_PER_MB
    // * this.partitionedRegion.rebalanceThreshold);
    maximumLocalBytes = (pr.getLocalMaxMemory() * PartitionedRegionHelper.BYTES_PER_MB);

    // this.bucketStats = new CachePerfStats(pr.getSystem(), "partition-" + pr.getName());
    bucketStats =
        new RegionPerfStats(pr.getCache().getInternalDistributedSystem().getStatisticsManager(),
            "RegionStats-partition-" + pr.getName(), pr.getCachePerfStats(), pr,
            pr.getCache().getMeterRegistry(), statisticsClock);
    keysOfInterest = new ConcurrentHashMap();
  }

  /**
   * This method creates a PartitionedRegionDataStore be invoking the PRDS Constructor.
   *
   * @return @throws PartitionedRegionException
   */
  static PartitionedRegionDataStore createDataStore(Cache cache, PartitionedRegion pr,
      PartitionAttributes pa, StatisticsClock statisticsClock) throws PartitionedRegionException {
    return new PartitionedRegionDataStore(pr, statisticsClock);
  }

  ConcurrentMap<Integer, BucketRegion> getLocalBucket2RegionMap() {
    return localBucket2RegionMap;
  }

  /**
   * Test to determine if this data store is managing a bucket
   *
   * @param bucketId the id of the bucket
   * @return true if the provided bucket is being managed
   */
  public boolean isManagingBucket(int bucketId) {
    BucketRegion buk = localBucket2RegionMap.get(bucketId);
    return buk != null && !buk.isDestroyed();
  }

  /**
   * Report the number of buckets currently managed by this DataStore
   */
  public short getBucketsManaged() {
    return (short) localBucket2RegionMap.size();
  }

  /**
   * Report an estimate of the number of primary buckets managed locally The result of this method
   * is not stable.
   */
  public int getNumberOfPrimaryBucketsManaged() {
    final AtomicInteger numPrimaries = new AtomicInteger();
    visitBuckets((bucketId, r) -> {
      BucketRegion br = (BucketRegion) r;
      if (br.getBucketAdvisor().isPrimary()) {
        numPrimaries.incrementAndGet();
      }
    });
    return numPrimaries.get();
  }

  public boolean isPartitionedRegionReady(PartitionedRegion partitionedRegion, final int bucketId) {
    List<PartitionedRegion> colocatedWithList = getColocatedChildRegions(partitionedRegion);
    if (colocatedWithList.size() == 0) {
      return partitionedRegion.isInitialized();
    }
    return areAllColocatedPartitionedRegionsReady(bucketId, colocatedWithList);
  }

  private boolean areAllColocatedPartitionedRegionsReady(int bucketId,
      List<PartitionedRegion> colocatedWithList) {
    return colocatedWithList.stream().allMatch(
        partitionedRegion -> isColocatedPartitionedRegionInitialized(partitionedRegion, bucketId));
  }

  private boolean isColocatedPartitionedRegionInitialized(PartitionedRegion partitionedRegion,
      final int bucketId) {
    if (!partitionedRegion.isInitialized()) {
      return false;
    }
    if (!partitionedRegion.getDataStore().isColocationComplete(bucketId)) {
      return false;
    }
    List<PartitionedRegion> colocatedWithList = getColocatedChildRegions(partitionedRegion);
    return areAllColocatedPartitionedRegionsReady(bucketId, colocatedWithList);
  }

  List<PartitionedRegion> getColocatedChildRegions(PartitionedRegion partitionedRegion) {
    return ColocationHelper.getColocatedChildRegions(partitionedRegion);
  }

  /**
   * Try to grab buckets for all the colocated regions /* In case we can't grab buckets there is no
   * going back
   *
   */

  protected CreateBucketResult grabFreeBucketRecursively(final int bucketId,
      final PartitionedRegion pr, final InternalDistributedMember moveSource,
      final boolean forceCreation, final boolean isRebalance, final boolean replaceOfflineData,
      final InternalDistributedMember creationRequestor, final boolean isDiskRecovery) {
    CreateBucketResult grab;
    DistributedMember dm = pr.getMyId();
    List<PartitionedRegion> colocatedWithList = ColocationHelper.getColocatedChildRegions(pr);
    // make sure we force creation and ignore redundancy checks for the child region.
    // if we created the parent bucket, we want to make sure we create the child bucket.
    grab = pr.getDataStore().grabFreeBucket(bucketId, dm, null, true, isRebalance, true,
        replaceOfflineData, creationRequestor);
    if (!grab.nowExists()) {
      if (logger.isDebugEnabled()) {
        logger.debug("Failed grab for bucketId = {}{}{}", pr.getPRId(),
            PartitionedRegion.BUCKET_ID_SEPARATOR,
            bucketId);
      }
      // Assert.assertTrue(nList.contains(partitionedRegion.getNode().getMemberId())
      // ,
      // " grab returned false and b2n does not contains this member.");
    }
    if (colocatedWithList != null) {
      for (final PartitionedRegion coLocatedWithPr : colocatedWithList) {
        if ((isDiskRecovery || coLocatedWithPr.isInitialized())
            && coLocatedWithPr.getDataStore().isColocationComplete(bucketId)) {
          grab = coLocatedWithPr.getDataStore().grabFreeBucketRecursively(bucketId, coLocatedWithPr,
              moveSource, forceCreation, isRebalance, replaceOfflineData, creationRequestor,
              isDiskRecovery);

          if (!grab.nowExists()) {
            if (logger.isDebugEnabled()) {
              logger.debug("Failed grab for bucketId = {}{}{}", pr.getPRId(),
                  PartitionedRegion.BUCKET_ID_SEPARATOR, bucketId);
            }
            // Assert.assertTrue(nList.contains(partitionedRegion.getNode().getMemberId())
            // ,
            // " grab returned false and b2n does not contains this member.");
          }
        }
      }
    }
    return grab;
  }

  /**
   * Attempts to map a bucket id to this node. Creates real storage for the bucket by adding a new
   * Region to bucket2Map. Bucket creation is done under the d-lock on b2n region.
   *
   * @param possiblyFreeBucketId the identity of the bucket + @param mustBeNew boolean enforcing
   *        that the bucket must not already exist
   * @param sender the member requesting the bucket
   * @param moveSource Where we are moving the bucket from, if this is a move.
   * @param forceCreation avoid any checks (with in reason) which might prevent bucket creation
   * @param isRebalance true if bucket creation is directed by rebalancing
   * @return true if successful
   */
  CreateBucketResult grabFreeBucket(final int possiblyFreeBucketId, final DistributedMember sender,
      final InternalDistributedMember moveSource, final boolean forceCreation,
      final boolean isRebalance, final boolean lockRedundancyLock, boolean replaceOffineData,
      InternalDistributedMember creationRequestor) {

    final boolean isDebugEnabled = logger.isDebugEnabled();

    long startTime = partitionedRegion.getPrStats().startBucketCreate(isRebalance);
    boolean createdBucket = false;
    PartitionedRegionObserver observer = PartitionedRegionObserverHolder.getInstance();
    observer.beforeBucketCreation(partitionedRegion, possiblyFreeBucketId);
    try {

      CreateBucketResult result = CreateBucketResult.FAILED;
      if (isManagingBucket(possiblyFreeBucketId)) {
        if (isDebugEnabled) {
          logger.debug("grabFreeBucket: VM {} already contains the bucket with bucketId={}{}{}",
              partitionedRegion.getMyId(), partitionedRegion.getPRId(),
              PartitionedRegion.BUCKET_ID_SEPARATOR, possiblyFreeBucketId);
        }
        partitionedRegion.checkReadiness();
        return CreateBucketResult.ALREADY_EXISTS;
      }
      StoppableReadLock parentBucketCreationLock = getParentBucketCreationLock();
      if (parentBucketCreationLock != null) {
        parentBucketCreationLock.lock();
      }

      try {
        if (!okToCreateChildBucket(possiblyFreeBucketId)) {
          return CreateBucketResult.FAILED;
        }
        StoppableReadLock lock = bucketCreationLock.readLock();
        lock.lock(); // prevent destruction while creating buckets
        try {
          // This counter is used by canAccomodateAnotherBucket to estimate if this member should
          // accept another bucket
          bucketCreatesInProgress.incrementAndGet();
          if (partitionedRegion.isDestroyed()) {
            return CreateBucketResult.FAILED;
          }

          // final boolean needsAllocation;
          if (isDebugEnabled) {
            logger.debug("grabFreeBucket: node list {} for bucketId={}{}{}",
                PartitionedRegionHelper.printCollection(partitionedRegion.getRegionAdvisor()
                    .getBucketOwners(possiblyFreeBucketId)),
                partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                possiblyFreeBucketId);
          }


          // Final accommodation check under synchronization for
          // a stable view of bucket creation
          if (!forceCreation) {
            // Currently the balancing check should only run when creating buckets,
            // as opposed to during bucket recovery... it is assumed that
            // ifRedudnancyNotSatisfied is false during bucket recovery
            if (!canAccommodateAnotherBucket()) {
              result = CreateBucketResult.FAILED;
              return result;
            }
          }

          ProxyBucketRegion buk =
              partitionedRegion.getRegionAdvisor().getProxyBucketArray()[possiblyFreeBucketId];
          // Prevent other threads on the same VM from creating this bucket.
          // It doesn't look the the redundancy lock actually correctly
          // handles multiple threads, and the isManagingBucket call
          // also needs to be done under this lock
          synchronized (buk) {
            // DAN - this just needs to be done holding a lock for this particular bucket
            if (!verifyBucketBeforeGrabbing(possiblyFreeBucketId)) {
              result = CreateBucketResult.FAILED;
              return result;
            }

            if (!isManagingBucket(possiblyFreeBucketId)) {
              Integer possiblyFreeBucketIdInt = possiblyFreeBucketId;

              BucketRegion bukReg = null;
              Object redundancyLock =
                  lockRedundancyLock(moveSource, possiblyFreeBucketId, replaceOffineData);
              // DAN - I hope this is ok to do without that bucket admin lock
              try {
                buk.initializePrimaryElector(creationRequestor);
                if (getPartitionedRegion().getColocatedWith() == null) {
                  buk.getBucketAdvisor().markAllShadowBucketsAsNonDestroyed();
                }
                if (getPartitionedRegion().isShadowPR()) {
                  getPartitionedRegion().getColocatedWithRegion().getRegionAdvisor()
                      .getBucketAdvisor(possiblyFreeBucketId).markAllShadowBucketsAsNonDestroyed();
                }
                bukReg = createBucketRegion(possiblyFreeBucketId);
                // Mark the bucket as hosting and distribute to peers
                // before we release the dlock. This makes sure that our peers
                // won't think they need to satisfy redundancy
                if (bukReg != null) {
                  // Let the data store know about the real bucket at this point
                  // so that other VMs which discover the real bucket via a
                  // profile exchange can send messages to the data store and
                  // safely use the bucket.
                  observer.beforeAssignBucket(partitionedRegion, possiblyFreeBucketId);
                  assignBucketRegion(bukReg.getId(), bukReg);
                  buk.setHosting(true);
                  bukReg.invokePartitionListenerAfterBucketCreated();
                } else {
                  if (buk.getPartitionedRegion().getColocatedWith() == null) {
                    buk.getBucketAdvisor().markAllShadowBucketsAsDestroyed();
                    // clear tempQueue for all the shadowPR buckets
                    clearAllTempQueueForShadowPR(buk.getBucketId());
                  }
                }
              } finally {
                releaseRedundancyLock(redundancyLock);
                if (bukReg == null) {
                  buk.clearPrimaryElector();
                }
              }

              if (bukReg != null) {
                if (isDebugEnabled) {
                  logger.debug("grabFreeBucket: mapped bucketId={}{}{} on node = {}",
                      partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                      possiblyFreeBucketId, partitionedRegion.getMyId());
                }

                createdBucket = true;
                result = CreateBucketResult.CREATED;
              } else {
                Assert.assertTrue(localBucket2RegionMap.get(possiblyFreeBucketIdInt) == null);
                result = CreateBucketResult.FAILED;
              }
            } else {
              // Handle the case where another concurrent thread decided to manage
              // the bucket and the creator may have died
              if (isDebugEnabled) {
                logger.debug("grabFreeBucket: bucketId={}{}{} already mapped on VM = {}",
                    partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                    possiblyFreeBucketId, partitionedRegion.getMyId());
              }
              result = CreateBucketResult.ALREADY_EXISTS;
            }
            if (isDebugEnabled) {
              logger.debug("grabFreeBucket: Mapped bucketId={}{}{}",
                  partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                  possiblyFreeBucketId);
            }
          }
        } catch (RegionDestroyedException rde) {
          RegionDestroyedException rde2 =
              new RegionDestroyedException(toString(), partitionedRegion.getFullPath(), rde);
          throw rde2;
        } catch (RedundancyAlreadyMetException e) {
          if (isDebugEnabled) {
            logger.debug("Redundancy already met {}{}{} assignment {}",
                partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                possiblyFreeBucketId,
                localBucket2RegionMap.get(possiblyFreeBucketId));
          }
          result = CreateBucketResult.REDUNDANCY_ALREADY_SATISFIED;
        } finally {
          bucketCreatesInProgress.decrementAndGet();
          lock.unlock(); // prevent destruction while creating buckets
        }
      } finally {
        if (parentBucketCreationLock != null) {
          parentBucketCreationLock.unlock();
        }
      }
      partitionedRegion.checkReadiness();
      partitionedRegion.checkClosed();

      return result;

    } catch (PartitionOfflineException validationException) {
      // GEODE-3055
      PartitionedRegion leader = ColocationHelper.getLeaderRegion(partitionedRegion);
      boolean isLeader = leader.equals(partitionedRegion);
      if (!isLeader) {
        leader.getDataStore().removeBucket(possiblyFreeBucketId, true);
        if (isDebugEnabled) {
          logger.debug("For bucket " + possiblyFreeBucketId
              + ", failed to create cololcated child bucket for "
              + partitionedRegion.getFullPath() + ", removed leader region "
              + leader.getFullPath() + " bucket.");
        }
      }
      throw validationException;
    } finally {
      partitionedRegion.getPrStats().endBucketCreate(startTime, createdBucket, isRebalance);
    }
  }

  protected void clearAllTempQueueForShadowPR(final int bucketId) {
    List<PartitionedRegion> colocatedWithList =
        ColocationHelper.getColocatedChildRegions(partitionedRegion);
    for (PartitionedRegion childRegion : colocatedWithList) {
      if (childRegion.isShadowPR()) {
        AbstractGatewaySender sender = childRegion.getParallelGatewaySender();
        if (sender == null) {
          return;
        }
        AbstractGatewaySenderEventProcessor eventProcessor = sender.getEventProcessor();
        if (eventProcessor == null) {
          return;
        }

        ConcurrentParallelGatewaySenderQueue queue =
            (ConcurrentParallelGatewaySenderQueue) eventProcessor.getQueue();

        if (queue == null) {
          return;
        }
        BlockingQueue<GatewaySenderEventImpl> tempQueue = queue.getBucketTmpQueue(bucketId);
        if (tempQueue != null) {
          synchronized (tempQueue) {
            for (GatewaySenderEventImpl event : tempQueue) {
              event.release();
            }
            tempQueue.clear();
          }
        }
      }
    }
  }

  public Object lockRedundancyLock(InternalDistributedMember moveSource, int bucketId,
      boolean replaceOffineData) {
    // TODO prperist - Make this thing easier to find!
    final PartitionedRegion.BucketLock bl = partitionedRegion.getRegionAdvisor()
        .getBucketAdvisor(bucketId).getProxyBucketRegion().getBucketLock();
    bl.lock();
    boolean succeeded = false;
    try {
      ProxyBucketRegion buk = partitionedRegion.getRegionAdvisor().getProxyBucketArray()[bucketId];
      if (!buk.checkBucketRedundancyBeforeGrab(moveSource, replaceOffineData)) {
        if (logger.isDebugEnabled()) {
          logger.debug("Redundancy already satisfied. current owners:{}",
              partitionedRegion.getRegionAdvisor().getBucketOwners(bucketId));
        }
        throw new RedundancyAlreadyMetException();
      }
      succeeded = true;
    } finally {
      if (!succeeded) {
        bl.unlock();
      }
    }

    return bl;
  }

  public void releaseRedundancyLock(Object lock) {
    try {

    } finally {
      PartitionedRegion.BucketLock bl = (BucketLock) lock;
      bl.unlock();
    }
  }

  private StoppableReadLock getParentBucketCreationLock() {
    PartitionedRegion colocatedRegion = ColocationHelper.getColocatedRegion(partitionedRegion);
    StoppableReadLock parentLock = null;
    if (colocatedRegion != null) {
      parentLock = colocatedRegion.getDataStore().bucketCreationLock.readLock();
      return parentLock;
    }
    return null;
  }

  /**
   * Returns false if this region is colocated and parent bucket does not exist.
   *
   * @return true if ok to make bucket
   */
  private boolean okToCreateChildBucket(int bucketId) {
    PartitionedRegion colocatedRegion = ColocationHelper.getColocatedRegion(partitionedRegion);
    if (colocatedRegion != null && !colocatedRegion.getDataStore().isManagingBucket(bucketId)) {
      if (logger.isDebugEnabled()) {
        logger.debug("okToCreateChildBucket - we don't manage the parent bucket");
      }
      return false;
    }
    return isColocationComplete(bucketId);
  }

  boolean isColocationComplete(int bucketId) {

    if (!ColocationHelper.isColocationComplete(partitionedRegion)) {
      ProxyBucketRegion pb =
          partitionedRegion.getRegionAdvisor().getProxyBucketArray()[bucketId];
      BucketPersistenceAdvisor persistenceAdvisor = pb.getPersistenceAdvisor();

      // Don't worry about colocation if we're recovering a persistent
      // bucket. The bucket must have been properly colocated earlier.
      if (persistenceAdvisor != null && persistenceAdvisor.wasHosting()) {
        return true;
      }
      if (logger.isDebugEnabled()) {
        logger.debug("Colocation is incomplete");
      }
      return false;
    }
    return true;
  }



  /**
   * This method creates bucket regions, based on redundancy level. If redundancy level is: a) = 1
   * it creates a local region b) >1 it creates a distributed region
   *
   * @return @throws CacheException
   */
  private BucketRegion createBucketRegion(int bucketId) {
    partitionedRegion.checkReadiness();

    BucketAttributesFactory factory = new BucketAttributesFactory();

    boolean isPersistent = partitionedRegion.getDataPolicy().withPersistence();
    if (isPersistent) {
      factory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
      setDiskAttributes(factory);
    } else {
      factory.setDataPolicy(DataPolicy.REPLICATE);
    }

    if (PartitionedRegion.DISABLE_SECONDARY_BUCKET_ACK) {
      factory.setScope(Scope.DISTRIBUTED_NO_ACK);
    } else {
      factory.setScope(Scope.DISTRIBUTED_ACK);
    }
    factory.setConcurrencyChecksEnabled(partitionedRegion.getConcurrencyChecksEnabled());
    factory.setIndexMaintenanceSynchronous(partitionedRegion.getIndexMaintenanceSynchronous());

    if (partitionedRegion.getValueConstraint() != null) {
      factory.setValueConstraint(partitionedRegion.getValueConstraint());
    }
    if (loader != null) {
      factory.setCacheLoader(loader);
    }
    factory.setEnableAsyncConflation(true);

    if (Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "PRDebug")) {
      factory.addCacheListener(createDebugBucketListener());
    }

    if (partitionedRegion.getStatisticsEnabled()) {
      factory.setStatisticsEnabled(true);
    }

    factory.setCloningEnabled(partitionedRegion.getCloningEnabled());

    ExpirationAttributes ea = partitionedRegion.getAttributes().getEntryIdleTimeout();
    if (ea != null) {
      factory.setEntryIdleTimeout(ea);
    }
    ea = partitionedRegion.getAttributes().getEntryTimeToLive();
    if (ea != null) {
      factory.setEntryTimeToLive(ea);
    }
    ea = partitionedRegion.getAttributes().getRegionIdleTimeout();
    if (ea != null) {
      if (ea.getAction() != ExpirationAction.DESTROY) {
        factory.setRegionIdleTimeout(ea);
      }
    }
    ea = partitionedRegion.getAttributes().getRegionTimeToLive();
    if (ea != null) {
      if (ea.getAction() != ExpirationAction.DESTROY) {
        factory.setRegionTimeToLive(ea);
      }
    }
    CustomExpiry ce = partitionedRegion.getAttributes().getCustomEntryIdleTimeout();
    if (ce != null) {
      factory.setCustomEntryIdleTimeout(ce);
    }
    ce = partitionedRegion.getAttributes().getCustomEntryTimeToLive();
    if (ce != null) {
      factory.setCustomEntryTimeToLive(ce);
    }

    if (partitionedRegion.getStatisticsEnabled()) {
      factory.setStatisticsEnabled(true);
    }
    EvictionAttributesImpl eva =
        (EvictionAttributesImpl) partitionedRegion.getEvictionAttributes();
    if (eva != null) {
      EvictionAttributes evBucket = eva;
      factory.setEvictionAttributes(evBucket);
      if (evBucket.getAction().isOverflowToDisk()) {
        setDiskAttributes(factory);
      }
    }

    factory.setCompressor(partitionedRegion.getCompressor());
    factory.setOffHeap(partitionedRegion.getOffHeap());

    factory.setBucketRegion(true); // prevent validation problems
    RegionAttributes attributes = factory.create();

    final String bucketRegionName = partitionedRegion.getBucketName(bucketId);

    LocalRegion rootRegion = PartitionedRegionHelper.getPRRoot(partitionedRegion.getCache());
    BucketRegion bucketRegion = null;

    if (Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "PRDebug")) {
      logger.info("createBucketRegion: Creating bucketId, {} name, {}.",
          partitionedRegion.bucketStringForLogs(bucketId),
          bucketRegionName);
    }
    try {
      final Bucket proxyBucket = partitionedRegion.getRegionAdvisor().getBucket(bucketId);
