/*
 * 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 java.lang.System.lineSeparator;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.toSet;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;

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

import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheStatistics;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.InterestPolicy;
import org.apache.geode.cache.InterestRegistrationEvent;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.PartitionedRegionDistributionException;
import org.apache.geode.cache.PartitionedRegionStorageException;
import org.apache.geode.cache.Region;
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.RegionMembershipListener;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionDataNotColocatedException;
import org.apache.geode.cache.TransactionDataRebalancedException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.client.internal.ClientMetadataService;
import org.apache.geode.cache.execute.EmptyRegionFunctionException;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.partition.PartitionListener;
import org.apache.geode.cache.partition.PartitionNotAvailableException;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.cache.persistence.PersistentID;
import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.IndexCreationException;
import org.apache.geode.cache.query.IndexExistsException;
import org.apache.geode.cache.query.IndexInvalidException;
import org.apache.geode.cache.query.IndexNameConflictException;
import org.apache.geode.cache.query.IndexType;
import org.apache.geode.cache.query.MultiIndexCreationException;
import org.apache.geode.cache.query.NameResolutionException;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.QueryInvocationTargetException;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.TypeMismatchException;
import org.apache.geode.cache.query.internal.Bag;
import org.apache.geode.cache.query.internal.CompiledSelect;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.cache.query.internal.QCompiler;
import org.apache.geode.cache.query.internal.QueryExecutor;
import org.apache.geode.cache.query.internal.ResultsBag;
import org.apache.geode.cache.query.internal.ResultsCollectionWrapper;
import org.apache.geode.cache.query.internal.ResultsSet;
import org.apache.geode.cache.query.internal.index.AbstractIndex;
import org.apache.geode.cache.query.internal.index.IndexCreationData;
import org.apache.geode.cache.query.internal.index.IndexManager;
import org.apache.geode.cache.query.internal.index.IndexUtils;
import org.apache.geode.cache.query.internal.index.PartitionedIndex;
import org.apache.geode.cache.query.internal.types.ObjectTypeImpl;
import org.apache.geode.cache.query.types.ObjectType;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LockServiceDestroyedException;
import org.apache.geode.distributed.internal.DistributionAdvisee;
import org.apache.geode.distributed.internal.DistributionAdvisor;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.FunctionExecutionPooledExecutor;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalDistributedSystem.DisconnectListener;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.OperationExecutors;
import org.apache.geode.distributed.internal.ProfileListener;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.locks.DLockRemoteToken;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.MemberDataBuilder;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.NanoTimer;
import org.apache.geode.internal.cache.BucketAdvisor.ServerBucketProfile;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.DestroyPartitionedRegionMessage.DestroyPartitionedRegionResponse;
import org.apache.geode.internal.cache.PutAllPartialResultException.PutAllPartialResult;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.eviction.EvictionController;
import org.apache.geode.internal.cache.eviction.HeapEvictor;
import org.apache.geode.internal.cache.execute.AbstractExecution;
import org.apache.geode.internal.cache.execute.BucketMovedException;
import org.apache.geode.internal.cache.execute.FunctionExecutionNodePruner;
import org.apache.geode.internal.cache.execute.FunctionRemoteContext;
import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException;
import org.apache.geode.internal.cache.execute.LocalResultCollector;
import org.apache.geode.internal.cache.execute.PartitionedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.PartitionedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.PartitionedRegionFunctionResultWaiter;
import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
import org.apache.geode.internal.cache.execute.ServerToClientFunctionResultSender;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.partitioned.ContainsKeyValueMessage;
import org.apache.geode.internal.cache.partitioned.ContainsKeyValueMessage.ContainsKeyValueResponse;
import org.apache.geode.internal.cache.partitioned.DestroyMessage;
import org.apache.geode.internal.cache.partitioned.DestroyMessage.DestroyResponse;
import org.apache.geode.internal.cache.partitioned.DumpAllPRConfigMessage;
import org.apache.geode.internal.cache.partitioned.DumpB2NRegion;
import org.apache.geode.internal.cache.partitioned.DumpB2NRegion.DumpB2NResponse;
import org.apache.geode.internal.cache.partitioned.DumpBucketsMessage;
import org.apache.geode.internal.cache.partitioned.FetchBulkEntriesMessage;
import org.apache.geode.internal.cache.partitioned.FetchBulkEntriesMessage.FetchBulkEntriesResponse;
import org.apache.geode.internal.cache.partitioned.FetchEntriesMessage;
import org.apache.geode.internal.cache.partitioned.FetchEntriesMessage.FetchEntriesResponse;
import org.apache.geode.internal.cache.partitioned.FetchEntryMessage;
import org.apache.geode.internal.cache.partitioned.FetchEntryMessage.FetchEntryResponse;
import org.apache.geode.internal.cache.partitioned.FetchKeysMessage;
import org.apache.geode.internal.cache.partitioned.FetchKeysMessage.FetchKeysResponse;
import org.apache.geode.internal.cache.partitioned.GetMessage;
import org.apache.geode.internal.cache.partitioned.GetMessage.GetResponse;
import org.apache.geode.internal.cache.partitioned.IdentityRequestMessage;
import org.apache.geode.internal.cache.partitioned.IdentityRequestMessage.IdentityResponse;
import org.apache.geode.internal.cache.partitioned.IdentityUpdateMessage;
import org.apache.geode.internal.cache.partitioned.IdentityUpdateMessage.IdentityUpdateResponse;
import org.apache.geode.internal.cache.partitioned.IndexCreationMsg;
import org.apache.geode.internal.cache.partitioned.InterestEventMessage;
import org.apache.geode.internal.cache.partitioned.InterestEventMessage.InterestEventResponse;
import org.apache.geode.internal.cache.partitioned.InvalidateMessage;
import org.apache.geode.internal.cache.partitioned.InvalidateMessage.InvalidateResponse;
import org.apache.geode.internal.cache.partitioned.PREntriesIterator;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.partitioned.PRSanityCheckMessage;
import org.apache.geode.internal.cache.partitioned.PRUpdateEntryVersionMessage;
import org.apache.geode.internal.cache.partitioned.PRUpdateEntryVersionMessage.UpdateEntryVersionResponse;
import org.apache.geode.internal.cache.partitioned.PartitionMessage.PartitionResponse;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionObserver;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionObserverHolder;
import org.apache.geode.internal.cache.partitioned.PutAllPRMessage;
import org.apache.geode.internal.cache.partitioned.PutMessage;
import org.apache.geode.internal.cache.partitioned.PutMessage.PutResult;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor.PartitionProfile;
import org.apache.geode.internal.cache.partitioned.RemoveAllPRMessage;
import org.apache.geode.internal.cache.partitioned.RemoveIndexesMessage;
import org.apache.geode.internal.cache.partitioned.SizeMessage;
import org.apache.geode.internal.cache.partitioned.SizeMessage.SizeResponse;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLogger;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory;
import org.apache.geode.internal.cache.persistence.PRPersistentConfig;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.BaseCommand;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.cache.wan.AbstractGatewaySenderEventProcessor;
import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderException;
import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.sequencelog.RegionLogger;
import org.apache.geode.internal.size.Sizeable;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.TransformUtils;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.logging.internal.executors.LoggingExecutors;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * A Region whose total storage is split into chunks of data (partitions) which are copied up to a
 * configurable level (for high availability) and placed on multiple VMs for improved performance
 * and increased storage capacity.
 */
public class PartitionedRegion extends LocalRegion
    implements CacheDistributionAdvisee, QueryExecutor {

  @Immutable
  public static final Random RANDOM =
      new Random(Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "PartitionedRegionRandomSeed",
          NanoTimer.getTime()));

  @MakeNotStatic
  private static final AtomicInteger SERIAL_NUMBER_GENERATOR = new AtomicInteger();

  /**
   * getNetworkHopType byte indicating this was the bucket owner for the last operation
   */
  public static final int NETWORK_HOP_NONE = 0;

  /**
   * getNetworkHopType byte indicating this was not the bucket owner and a message had to be sent to
   * a primary in the same server group
   */
  private static final int NETWORK_HOP_TO_SAME_GROUP = 1;

  /**
   * getNetworkHopType byte indicating this was not the bucket owner and a message had to be sent to
   * a primary in a different server group
   */
  public static final int NETWORK_HOP_TO_DIFFERENT_GROUP = 2;
  public static final String DATA_MOVED_BY_REBALANCE =
      "Transactional data moved, due to rebalancing.";

  private final DiskRegionStats diskRegionStats;

  /**
   * Changes scope of replication to secondary bucket to SCOPE.DISTRIBUTED_NO_ACK
   */
  static final boolean DISABLE_SECONDARY_BUCKET_ACK =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "disablePartitionedRegionBucketAck");

  /**
   * A debug flag used for testing calculation of starting bucket id
   */
  @MutableForTesting
  public static boolean BEFORE_CALCULATE_STARTING_BUCKET_FLAG = false;

  /**
   * Thread specific random number
   */
  private static final ThreadLocal threadRandom = new ThreadLocal() {
    @Override
    protected Object initialValue() {
      int i = RANDOM.nextInt();
      if (i < 0) {
        i = -1 * i;
      }
      return i;
    }
  };

  /**
   * Global Region for storing PR config ( PRName->PRConfig). This region would be used to resolve
   * PR name conflict.*
   */
  private volatile Region<String, PartitionRegionConfig> prRoot;

  /**
   *
   * PartitionedRegionDataStore class takes care of data storage for the PR. This will contain the
   * bucket Regions to store data entries for PR*
   */
  protected PartitionedRegionDataStore dataStore;

  /**
   * The advisor that hold information about this partitioned region
   */
  private final RegionAdvisor distAdvisor;
  private final SenderIdMonitor senderIdMonitor;

  /** Logging mechanism for debugging */
  private static final Logger logger = LogService.getLogger();

  /** cleanup flags * */
  private boolean cleanPRRegistration = false;

  /** Time to wait for for acquiring distributed lock ownership */
  private static final long VM_OWNERSHIP_WAIT_TIME = PRSystemPropertyGetter.parseLong(
      System.getProperty(PartitionedRegionHelper.VM_OWNERSHIP_WAIT_TIME_PROPERTY),
      PartitionedRegionHelper.VM_OWNERSHIP_WAIT_TIME_DEFAULT);

  /**
   * default redundancy level is 0.
   */
  final int redundantCopies;

  /**
   * The miminum amount of redundancy needed for a write operation
   */
  final int minimumWriteRedundancy;

  /**
   * The miminum amount of redundancy needed for a read operation
   */
  final int minimumReadRedundancy;

  /**
   * Ratio of currently allocated memory to maxMemory that triggers rebalance activity.
   */
  static final float rebalanceThreshold = 0.75f;

  /** The maximum memory allocated for this node in Mb */
  final int localMaxMemory;

  /** The maximum milliseconds for retrying operations */
  private final int retryTimeout;

  /**
   * The statistics for this PR
   */
  public final PartitionedRegionStats prStats;

  // private Random random = new Random(System.currentTimeMillis());

  /** Number of initial buckets */
  private final int totalNumberOfBuckets;

  // private static final boolean throwIfNoNodesLeft = true;

  public static final int DEFAULT_RETRY_ITERATIONS = 3;

  /**
   * Flag to indicate if a cache loader is present
   */
  private volatile boolean haveCacheLoader;

  /**
   * Region identifier used for DLocks (Bucket and Region)
   */
  private final String regionIdentifier;

  /**
   * Maps each PR to a prId. This prId will uniquely identify the PR.
   */
  @MakeNotStatic
  private static final PRIdMap prIdToPR = new PRIdMap();

  /**
   * Flag to indicate whether region is closed
   *
   */
  public volatile boolean isClosed = false;

  /**
   * a flag indicating that the PR is destroyed in this VM
   */
  public volatile boolean isLocallyDestroyed = false;

  /**
   * the thread locally destroying this pr. not volatile, so always check isLocallyDestroyed before
   * checking locallyDestroyingThread
   *
   * Concurrency: {@link #isLocallyDestroyed} is volatile
   */
  Thread locallyDestroyingThread;

  // TODO someone please add a javadoc for this
  private volatile boolean hasPartitionedIndex = false;

  /**
   * regionMembershipListener notification requires this to be plugged into a PR's RegionAdvisor
   */
  private final AdvisorListener advisorListener = new AdvisorListener();

  /*
   * Map containing <IndexTask, FutureTask<IndexTask> or Index>. IndexTask represents an index thats
   * completely created or one thats in create phase. This is done in order to avoid synchronization
   * on the indexes.
   */
  private final ConcurrentMap indexes = new ConcurrentHashMap();

  private volatile boolean recoveredFromDisk;

  public static final int RUNNING_MODE = -1;
  public static final int PRIMARY_BUCKETS_LOCKED = 1;
  public static final int DISK_STORE_FLUSHED = 2;
  public static final int OFFLINE_EQUAL_PERSISTED = 3;

  private volatile int shutDownAllStatus = RUNNING_MODE;

  private final long birthTime = System.currentTimeMillis();

  public void setShutDownAllStatus(int newStatus) {
    shutDownAllStatus = newStatus;
  }

  private final PartitionedRegion colocatedWithRegion;

  private final ColocationLoggerFactory colocationLoggerFactory;

  private final AtomicReference<ColocationLogger> missingColocatedRegionLogger =
      new AtomicReference<>();

  private List<BucketRegion> sortedBuckets;

  private ScheduledExecutorService bucketSorter;

  private final ConcurrentMap<String, Integer[]> partitionsMap = new ConcurrentHashMap<>();

  public ConcurrentMap<String, Integer[]> getPartitionsMap() {
    return partitionsMap;
  }

  /**
   * for wan shadowPR
   */
  private boolean enableConflation;

  private final Object indexLock = new Object();

  private final Set<ColocationListener> colocationListeners = ConcurrentHashMap.newKeySet();

  public void addColocationListener(ColocationListener colocationListener) {
    if (colocationListener != null) {
      colocationListeners.add(colocationListener);
    }
  }

  public void removeColocationListener(ColocationListener colocationListener) {
    colocationListeners.remove(colocationListener);
  }

  ScheduledExecutorService getBucketSorter() {
    return bucketSorter;
  }

  @VisibleForTesting
  public static PRIdMap getPrIdToPR() {
    return prIdToPR;
  }

  /**
   * Byte 0 = no NWHOP Byte 1 = NWHOP to servers in same server-grp Byte 2 = NWHOP tp servers in
   * other server-grp
   */
  private static final ThreadLocal<Byte> networkHopType = new ThreadLocal<Byte>() {
    @Override
    protected Byte initialValue() {
      return (byte) NETWORK_HOP_NONE;
    }
  };

  public void clearNetworkHopData() {
    networkHopType.remove();
    metadataVersion.remove();
  }

  private void setNetworkHopType(Byte value) {
    networkHopType.set(value);
  }

  /**
   * If the last operation in the current thread required a one-hop to another server who held the
   * primary bucket for the operation then this will return something other than NETWORK_HOP_NONE.
   * <p>
   * see NETWORK_HOP_NONE, NETWORK_HOP_TO_SAME_GROUP and NETWORK_HOP_TO_DIFFERENT_GROUP
   */
  public byte getNetworkHopType() {
    return networkHopType.get();
  }

  private static final ThreadLocal<Byte> metadataVersion = new ThreadLocal<Byte>() {
    @Override
    protected Byte initialValue() {
      return ClientMetadataService.INITIAL_VERSION;
    }
  };

  private void setMetadataVersion(Byte value) {
    metadataVersion.set(value);
  }

  public byte getMetadataVersion() {
    return metadataVersion.get();
  }

  /**
   * Returns the EvictionController for this PR. This is needed to find the single instance of
   * EvictionController created early for a PR when it is recovered from disk. This fixes bug 41938
   */
  public EvictionController getPREvictionControllerFromDiskInitialization() {
    EvictionController result = null;
    if (getDiskStore() != null) {
      result = getDiskStore().getExistingPREvictionContoller(this);
    }
    return result;
  }


  public void clearPREvictionControllerFromDiskInitialization() {
    if (getDiskStore() != null) {
      getDiskStore().clearExistingPREvictionContoller(this);
    }
  }



  @Override
  public boolean remove(Object key, Object value, Object callbackArg) {
    final long startTime = prStats.getTime();
    try {
      return super.remove(key, value, callbackArg);
    } finally {
      prStats.endDestroy(startTime);
    }
  }

  public PartitionListener[] getPartitionListeners() {
    return partitionListeners;
  }

  /**
   * Return canonical representation for a bucket (for logging)
   *
   * @param bucketId the bucket
   * @return a String representing this PR and the bucket
   */
  public String bucketStringForLogs(int bucketId) {
    return getPRId() + BUCKET_ID_SEPARATOR + bucketId;
  }

  /** Separator between PRId and bucketId for creating bucketString */
  public static final String BUCKET_ID_SEPARATOR = ":";

  /**
   * Clear the prIdMap, typically used when disconnecting from the distributed system or clearing
   * the cache
   */
  public static void clearPRIdMap() {
    synchronized (prIdToPR) {
      prIdToPR.clear();
    }
  }

  @Immutable
  private static final DisconnectListener dsPRIdCleanUpListener = new DisconnectListener() {
    @Override
    public String toString() {
      return "Shutdown listener for PartitionedRegion";
    }

    @Override
    public void onDisconnect(InternalDistributedSystem sys) {
      clearPRIdMap();
    }
  };

  PartitionedRegionRedundancyTracker getRedundancyTracker() {
    return redundancyTracker;
  }

  public <T> T computeWithPrimaryLocked(Object key, Callable<T> callable) throws Exception {
    int bucketId = PartitionedRegionHelper.getHashKey(this, null, key, null, null);

    BucketRegion br;
    try {
      br = dataStore.getInitializedBucketForId(key, bucketId);
    } catch (ForceReattemptException e) {
      throw new BucketMovedException(e);
    }

    try {
      br.doLockForPrimary(false);
    } catch (PrimaryBucketException e) {
      throw new PrimaryBucketLockException("retry since primary lock failed: " + e);
    }

    try {
      return callable.call();
    } finally {
      br.doUnlockForPrimary();
    }
  }


  public static class PRIdMap extends HashMap {
    private static final long serialVersionUID = 3667357372967498179L;
    public static final String DESTROYED = "Partitioned Region Destroyed";

    static final String LOCALLY_DESTROYED = "Partitioned Region Is Locally Destroyed";

    static final String FAILED_REGISTRATION = "Partitioned Region's Registration Failed";

    public static final String NO_PATH_FOUND = "NoPathFound";

    private volatile boolean cleared = true;

    @Override
    public Object get(Object key) {
      throw new UnsupportedOperationException(
          "PRIdMap.get not supported - use getRegion instead");
    }

    public Object getRegion(Object key) throws PRLocallyDestroyedException {
      Assert.assertTrue(key instanceof Integer);

      Object o = super.get(key);
      if (o == DESTROYED) {
        throw new RegionDestroyedException(
            String.format("Region for prId= %s is destroyed",
                key),
            NO_PATH_FOUND);
      }
      if (o == LOCALLY_DESTROYED) {
        throw new PRLocallyDestroyedException(
            String.format("Region with prId= %s is locally destroyed on this node",
                key));
      }
      if (o == FAILED_REGISTRATION) {
        throw new PRLocallyDestroyedException(
            String.format("Region with prId= %s failed initialization on this node",
                key));
      }
      return o;
    }

    @Override
    public Object remove(final Object key) {
      return put(key, DESTROYED, true);
    }

    @Override
    public Object put(final Object key, final Object value) {
      return put(key, value, true);
    }

    public Object put(final Object key, final Object value, boolean sendIdentityRequestMessage) {
      if (cleared) {
        cleared = false;
      }

      if (key == null) {
        throw new NullPointerException(
            "null key not allowed for prIdToPR Map");
      }
      if (value == null) {
        throw new NullPointerException(
            "null value not allowed for prIdToPR Map");
      }
      Assert.assertTrue(key instanceof Integer);
      if (sendIdentityRequestMessage) {
        IdentityRequestMessage.setLatestId((Integer) key);
      }
      if ((super.get(key) == DESTROYED) && (value instanceof PartitionedRegion)) {
        throw new PartitionedRegionException(
            String.format("Can NOT reuse old Partitioned Region Id= %s",
                key));
      }
      return super.put(key, value);
    }

    @Override
    public void clear() {
      cleared = true;
      super.clear();
    }

    public synchronized String dump() {
      StringBuilder sb = new StringBuilder("prIdToPR Map@");
      sb.append(System.identityHashCode(prIdToPR)).append(':').append(lineSeparator());
      Map.Entry mapEntry;
      for (Iterator iterator = prIdToPR.entrySet().iterator(); iterator.hasNext();) {
        mapEntry = (Map.Entry) iterator.next();
        sb.append(mapEntry.getKey()).append("=>").append(mapEntry.getValue());
        if (iterator.hasNext()) {
          sb.append(lineSeparator());
        }
      }
      return sb.toString();
    }
  }

  private int partitionedRegionId = -3;

  /** Node description */
  private final Node node;

  /** Helper Object for redundancy Management of PartitionedRegion */
  private final PRHARedundancyProvider redundancyProvider;

  /**
   * flag saying whether this VM needs cache operation notifications from other members
   */
  private boolean requiresNotification;

  /**
   * Latch that signals when the Bucket meta-data is ready to receive updates
   */
  private final StoppableCountDownLatch initializationLatchAfterBucketIntialization;

  public static final String RETRY_TIMEOUT_PROPERTY =
      GeodeGlossary.GEMFIRE_PREFIX + "partitionedRegionRetryTimeout";

  private final PartitionRegionConfigValidator validator;

  final List<FixedPartitionAttributesImpl> fixedPAttrs;

  private byte fixedPASet;

  private final List<PartitionedRegion> colocatedByList =
      new CopyOnWriteArrayList<>();

  private final PartitionListener[] partitionListeners;

  private boolean isShadowPR = false;

  private AbstractGatewaySender parallelGatewaySender = null;

  private final PartitionedRegionRedundancyTracker redundancyTracker;

  private boolean regionCreationNotified;

  private boolean sentGatewaySenderStoppedMessage = false;

  public interface RegionAdvisorFactory {
    RegionAdvisor create(PartitionedRegion region);
  }

  /**
   * Constructor for a PartitionedRegion. This has an accessor (Region API) functionality and
   * contains a datastore for actual storage. An accessor can act as a local cache by having a local
   * storage enabled. A PartitionedRegion can be created by a factory method of RegionFactory.java
   * and also by invoking Cache.createRegion(). (Cache.xml etc to be added)
   */
  public PartitionedRegion(String regionName,
      RegionAttributes regionAttributes,
      LocalRegion parentRegion,
      InternalCache cache,
      InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock,
      ColocationLoggerFactory colocationLoggerFactory) {
    this(regionName, regionAttributes, parentRegion, cache, internalRegionArgs, statisticsClock,
        colocationLoggerFactory,
        region -> RegionAdvisor.createRegionAdvisor(region));
  }

  PartitionedRegion(String regionName,
      RegionAttributes regionAttributes,
      LocalRegion parentRegion,
      InternalCache cache,
      InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock,
      ColocationLoggerFactory colocationLoggerFactory,
      RegionAdvisorFactory regionAdvisorFactory) {

    super(regionName, regionAttributes, parentRegion, cache, internalRegionArgs,
        new PartitionedRegionDataView(), statisticsClock);

    this.colocationLoggerFactory = colocationLoggerFactory;
    node = initializeNode();
    prStats = new PartitionedRegionStats(cache.getDistributedSystem(), getFullPath(),
        statisticsClock);
    regionIdentifier = getFullPath().replace(Region.SEPARATOR_CHAR, '#');

