/*
 * 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 org.apache.geode.internal.cache.LocalRegion.InitializationLevel.AFTER_INITIAL_IMAGE;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE;
import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_NEW_VALUE;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;

import jakarta.transaction.RollbackException;
import jakarta.transaction.Status;
import jakarta.transaction.SystemException;
import jakarta.transaction.Transaction;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.CopyHelper;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.Delta;
import org.apache.geode.DeltaSerializationException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.LogWriter;
import org.apache.geode.Statistics;
import org.apache.geode.SystemFailure;
import org.apache.geode.admin.internal.SystemMemberCacheEventProcessor;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheEvent;
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.CacheRuntimeException;
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.DiskStoreFactory;
import org.apache.geode.cache.DiskWriteAttributes;
import org.apache.geode.cache.DiskWriteAttributesFactory;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.EntryExistsException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.FailedSynchronizationException;
import org.apache.geode.cache.InterestRegistrationEvent;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.Operation;
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.RegionReinitializedException;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.ServerOperationException;
import org.apache.geode.cache.client.SubscriptionNotEnabledException;
import org.apache.geode.cache.client.internal.Connection;
import org.apache.geode.cache.client.internal.Endpoint;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.ServerRegionProxy;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.persistence.ConflictingPersistentDataException;
import org.apache.geode.cache.query.FunctionDomainException;
import org.apache.geode.cache.query.Index;
import org.apache.geode.cache.query.IndexExistsException;
import org.apache.geode.cache.query.IndexMaintenanceException;
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.QueryInvalidException;
import org.apache.geode.cache.query.QueryInvocationTargetException;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.TypeMismatchException;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.cache.query.internal.cq.CqService;
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.IndexProtocol;
import org.apache.geode.cache.query.internal.index.IndexUtils;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionAdvisor;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ResourceEvent;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.ClassLoadUtils;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.DiskInitFile.DiskRegionFlag;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.InitialImageOperation.GIIStatus;
import org.apache.geode.internal.cache.PutAllPartialResultException.PutAllPartialResult;
import org.apache.geode.internal.cache.RegionMap.ARMLockTestHook;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.control.MemoryThresholds;
import org.apache.geode.internal.cache.control.ResourceListener;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.event.NonDistributedEventTracker;
import org.apache.geode.internal.cache.eviction.EvictableEntry;
import org.apache.geode.internal.cache.eviction.EvictionController;
import org.apache.geode.internal.cache.eviction.EvictionCounters;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.LocalResultCollector;
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.Bucket;
import org.apache.geode.internal.cache.partitioned.RedundancyAlreadyMetException;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory;
import org.apache.geode.internal.cache.persistence.DefaultDiskDirs;
import org.apache.geode.internal.cache.persistence.DiskRegionView;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.query.IndexMap;
import org.apache.geode.internal.cache.persistence.query.mock.IndexMapImpl;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientHealthMonitor;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ClientTombstoneMessage;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessage;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.cache.versions.RegionVersionHolder;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
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.lang.SystemPropertyHelper;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.offheap.ReferenceCountHelper;
import org.apache.geode.internal.offheap.Releasable;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.CopyOnWriteHashMap;
import org.apache.geode.internal.util.concurrent.FutureResult;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.JSONFormatter;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implementation of a local scoped-region. Note that this class has a different meaning starting
 * with 3.0. In previous versions, a LocalRegion was the representation of a region in the VM.
 * Starting with 3.0, a LocalRegion is a non-distributed region. The subclass DistributedRegion adds
 * distribution behavior.
 */
public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
    ResourceListener<MemoryEvent>, InternalPersistentRegion {

  private static final Logger logger = LogService.getLogger();

  @Override
  public boolean isRegionInvalid() {
    return regionInvalid;
  }

  /**
   * Set to true after an invalidate region expiration so we don't get multiple expirations
   */
  @Override
  public void setRegionInvalid(boolean regionInvalid) {
    this.regionInvalid = regionInvalid;
  }

  /**
   * Prevents access to this region until it is done initializing, except for some special
   * initializing operations such as replying to create region messages In JDK 1.5 we will use
   * java.util.concurrent.CountDownLatch instead of org.apache.geode.internal.util.CountDownLatch.
   */
  @Override
  public StoppableCountDownLatch getInitializationLatchBeforeGetInitialImage() {
    return initializationLatchBeforeGetInitialImage;
  }

  @Override
  public StoppableCountDownLatch getInitializationLatchAfterGetInitialImage() {
    return initializationLatchAfterGetInitialImage;
  }

  // initialization level
  public enum InitializationLevel {
    AFTER_INITIAL_IMAGE, BEFORE_INITIAL_IMAGE, ANY_INIT
  }

  /**
   * thread local to indicate that this thread should bypass the initialization Latch
   */
  private static final ThreadLocal<InitializationLevel> initializationThread =
      ThreadLocal.withInitial(() -> AFTER_INITIAL_IMAGE);

  private Object regionUserAttribute;

  private final Map<Object, Object> entryUserAttributes = new ConcurrentHashMap<>();

  private final String regionName;

  private final LocalRegion parentRegion;

  /**
   * set to true only if isDestroyed is also true and region is about to be recreated due to
   * reinitialization by loading of a snapshot, etc.
   */
  private volatile boolean reinitialized_old;

  protected volatile boolean isDestroyed;

  /**
   * In case of parallel wan, when a destroy is called on userPR, it waits for parallelQueue to
   * drain and then destroys parallelQueue. In this time if operation like put happens on userPR
   * then it will keep on building parallel queue increasing time of userPR to get destroyed.this
   * volatile boolean will block such put operation by throwing RegionDestroyedException
   */
  volatile boolean isDestroyedForParallelWAN;

  /**
   * set to true after snapshot is loaded, to help get initial image make sure this is the right
   * incarnation of this region
   */
  private volatile boolean reinitialized_new;

  /**
   * Lock used to prevent multiple concurrent destroy region operations
   */
  private Semaphore destroyLock;

  /**
   * GuardedBy regionExpiryLock.
   */
  private RegionTTLExpiryTask regionTTLExpiryTask;

  /**
   * GuardedBy regionExpiryLock.
   */
  private RegionIdleExpiryTask regionIdleExpiryTask;

  private final Object regionExpiryLock = new Object();

  /**
   * GuardedBy regionExpiryLock. Keeps track of how many txs are writing to this region.
   */
  private int txRefCount;

  private volatile boolean regionInvalid;

  /**
   * TODO: make this private and introduce wrappers
   */
  public final RegionMap entries;

  /**
   * Set to true if this region supports transaction else false.
   */
  private final boolean supportsTX;

  /**
   * tracks region-level version information for members
   */
  private final RegionVersionVector versionVector;

  private static final Pattern[] QUERY_PATTERNS = new Pattern[] {
      Pattern.compile("^\\(*select .*",
          Pattern.CASE_INSENSITIVE | Pattern.UNICODE_CASE | Pattern.DOTALL),
      Pattern.compile("^import .*",
          Pattern.CASE_INSENSITIVE | Pattern.UNICODE_CASE | Pattern.DOTALL)};

  public static final String EXPIRY_MS_PROPERTY =
      GeodeGlossary.GEMFIRE_PREFIX + "EXPIRY_UNITS_MS";

  /**
   * Used by unit tests to set expiry to milliseconds instead of the default seconds.
   *
   * @since GemFire 5.0
   */
  private final boolean EXPIRY_UNITS_MS;

  @VisibleForTesting
  boolean isExpiryUnitsMilliseconds() {
    return EXPIRY_UNITS_MS;
  }

  private final EntryEventFactory entryEventFactory;
  private final RegionMapConstructor regionMapConstructor;

  /**
   * contains Regions themselves // marked volatile to make sure it is fully initialized before
   * being // accessed; (actually should be final)
   */
  private final ConcurrentMap subregions;

  private final Object subregionsLock = new Object();

  private final StoppableCountDownLatch initializationLatchBeforeGetInitialImage;

  private final StoppableCountDownLatch initializationLatchAfterGetInitialImage;

  /**
   * Used to hold off cache listener events until the afterRegionCreate is called
   *
   * @since GemFire 5.0
   */
  private final StoppableCountDownLatch afterRegionCreateEventLatch;

  /**
   * Set to true the first time isInitialized returns true.
   */
  private volatile boolean initialized;

  /**
   * Used for accessing region data on disk
   */
  private final DiskRegion diskRegion;

  /**
   * Used for serializing netSearch and netLoad on a per key basis. CM <Object, Future>
   */
  private final ConcurrentMap getFutures = new ConcurrentHashMap();

  /**
   * TODO: This boolean needs to be made true if the test needs to receive a synchronous callback
   * just after clear on map is done. Its visibility is default so that only tests present in
   * org.apache.geode.internal.cache will be able to see it
   */
  @MakeNotStatic("This is modified in production code")
  public static boolean ISSUE_CALLBACKS_TO_CACHE_OBSERVER;

  /**
   * A flag used to indicate that this Region is being used as an administrative Region, holding
   * meta-data for a PartitionedRegion
   */
  private final boolean isUsedForPartitionedRegionAdmin;

  private final boolean isUsedForPartitionedRegionBucket;

  private final boolean isUsedForMetaRegion;

  private final boolean isMetaRegionWithTransactions;

  private final boolean isUsedForSerialGatewaySenderQueue;

  private final boolean isUsedForParallelGatewaySenderQueue;

  private final AbstractGatewaySender serialGatewaySender;

  /**
   * The factory used to create the LoaderHelper when a loader is invoked
   */
  final LoaderHelperFactory loaderHelperFactory;

  /**
   * Allow for different CachePerfStats locations... primarily for PartitionedRegions
   */
  private final CachePerfStats cachePerfStats;

  private final boolean hasOwnStats;

  private final ImageState imageState;

  private final EventTracker eventTracker;

  /**
   * Register interest count to track if any register interest is in progress for this region. This
   * count will be incremented when register interest starts and decremented when register interest
   * finishes.
   * <p>
   * since always written while holding an exclusive write lock and only read while holding a read
   * lock it does not need to be atomic or protected by any other sync.
   * <p>
   * GuardedBy {@link #imageState}
   */
  private int riCnt;

  /**
   * Map of subregion full paths to serial numbers. These are subregions that were destroyed when
   * this region was destroyed. This map remains null until this region is destroyed.
   */
  private volatile HashMap destroyedSubregionSerialNumbers;

  /**
   * This boolean is true when a member who has this region is running low on memory. It is used to
   * reject region operations.
   */
  private final AtomicBoolean memoryThresholdReached = new AtomicBoolean(false);

  /**
   * Lock for updating PR MetaData on client side
   * <p>
   * TODO: move this to ClientMetadataService into {@code Map<Region, Lock>}
   */
  private final Lock clientMetaDataLock = new ReentrantLock();

  /**
   * Lock for updating the cache service profile for the region.
   */
  private final Lock cacheServiceProfileUpdateLock = new ReentrantLock();

  public void executeSynchronizedOperationOnCacheProfiles(Runnable operation) {
    cacheServiceProfileUpdateLock.lock();
    try {
      operation.run();
    } finally {
      cacheServiceProfileUpdateLock.unlock();
    }
  }

  private final CancelCriterion stopper = createStopper();

  private CancelCriterion createStopper() {
    return new Stopper();
  }

  private final TestCallable testCallable;

  /**
   * ThreadLocal used to set the current region being initialized.
   *
   * Currently used by the OpLog layer.
   */
  private static final ThreadLocal<LocalRegion> initializingRegion = new ThreadLocal<>();

  /**
   * Get the current initializing region as set in the ThreadLocal.
   *
   * Note that this value is cleared after the initialization of LocalRegion is done so is valid
   * only for the duration of region creation and initialization.
   */
  static LocalRegion getInitializingRegion() {
    return initializingRegion.get();
  }

  @Override
  public CancelCriterion getCancelCriterion() {
    return stopper;
  }

  private final CopyOnWriteHashMap<String, CacheServiceProfile> cacheServiceProfiles =
      new CopyOnWriteHashMap<>();

  private static String calcFullPath(String regionName, Region parentRegion) {
    StringBuilder buf;
    if (parentRegion == null) {
      buf = new StringBuilder(regionName.length() + 1);
    } else {
      String parentFull = parentRegion.getFullPath();
      buf = new StringBuilder(parentFull.length() + regionName.length() + 1);
      buf.append(parentFull);
    }
    buf.append(Region.SEPARATOR).append(regionName);
    return buf.toString();
  }

  protected LocalRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
      InternalCache cache, InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock) throws DiskAccessException {
    this(regionName, attrs, parentRegion, cache, internalRegionArgs, new LocalRegionDataView(),
        statisticsClock);
  }

  protected LocalRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
      InternalCache cache, InternalRegionArguments internalRegionArgs,
      InternalDataView internalDataView, StatisticsClock statisticsClock)
      throws DiskAccessException {
    this(regionName, attrs, parentRegion, cache, internalRegionArgs, internalDataView,
        RegionMapFactory::createVM, new DefaultServerRegionProxyConstructor(),
        new DefaultEntryEventFactory(), poolName -> (PoolImpl) PoolManager.find(poolName),
        (LocalRegion region) -> new RegionPerfStats(
            cache.getInternalDistributedSystem().getStatisticsManager(),
            "RegionStats-" + regionName, cache.getCachePerfStats(),
            region, cache.getMeterRegistry(), statisticsClock),
        statisticsClock);
  }

  @VisibleForTesting
  LocalRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
      InternalCache cache, InternalRegionArguments internalRegionArgs,
      InternalDataView internalDataView, RegionMapConstructor regionMapConstructor,
      ServerRegionProxyConstructor serverRegionProxyConstructor,
      EntryEventFactory entryEventFactory, PoolFinder poolFinder,
      java.util.function.Function<LocalRegion, RegionPerfStats> regionPerfStatsFactory,
      StatisticsClock statisticsClock)
      throws DiskAccessException {
    super(cache, attrs, regionName, internalRegionArgs, poolFinder, statisticsClock);

    this.regionMapConstructor = regionMapConstructor;
    this.entryEventFactory = entryEventFactory;

    EXPIRY_UNITS_MS = parentRegion != null ? parentRegion.EXPIRY_UNITS_MS
        : Boolean.getBoolean(EXPIRY_MS_PROPERTY);

    Assert.assertTrue(regionName != null, "regionName must not be null");
    sharedDataView = internalDataView;
    this.regionName = regionName;
    this.parentRegion = parentRegion;
    fullPath = calcFullPath(regionName, parentRegion);

    String myName = getFullPath();
    if (internalRegionArgs.getPartitionedRegion() != null) {
      myName = internalRegionArgs.getPartitionedRegion().getFullPath();
    }
    offHeap = attrs.getOffHeap() || Boolean.getBoolean(myName + ":OFF_HEAP");
    if (getOffHeap()) {
      if (cache.getOffHeapStore() == null) {
        throw new IllegalStateException(
            String.format(
                "The region %s was configured to use off heap memory but no off heap memory was configured",
                myName));
      }
    }

    initializationLatchBeforeGetInitialImage = new StoppableCountDownLatch(stopper, 1);
    initializationLatchAfterGetInitialImage = new StoppableCountDownLatch(stopper, 1);
    afterRegionCreateEventLatch = new StoppableCountDownLatch(stopper, 1);

    if (internalRegionArgs.getUserAttribute() != null) {
      setUserAttribute(internalRegionArgs.getUserAttribute());
    }
    initializingRegion.set(this);

    diskStoreImpl = findDiskStore(attrs, internalRegionArgs);
    diskRegion = createDiskRegion(internalRegionArgs);
    entries = createRegionMap(internalRegionArgs);
    subregions = new ConcurrentHashMap();

    if (internalRegionArgs.getCachePerfStatsHolder() != null) {
      HasCachePerfStats cachePerfStatsHolder = internalRegionArgs.getCachePerfStatsHolder();
      hasOwnStats = cachePerfStatsHolder.hasOwnStats();
      cachePerfStats = cachePerfStatsHolder.getCachePerfStats();
    } else {
      if (attrs.getPartitionAttributes() != null || isInternalRegion()
          || internalRegionArgs.isUsedForMetaRegion()) {
        hasOwnStats = false;
        cachePerfStats = cache.getCachePerfStats();
      } else {
        hasOwnStats = true;
        cachePerfStats = regionPerfStatsFactory.apply(this);
      }
    }

    // we only need a destroy lock if this is a root
    if (parentRegion == null) {
      initRoot();
    }

    if (internalRegionArgs.getLoaderHelperFactory() != null) {
      loaderHelperFactory = internalRegionArgs.getLoaderHelperFactory();
    } else {
      loaderHelperFactory = this;
    }

    isUsedForPartitionedRegionAdmin = internalRegionArgs.isUsedForPartitionedRegionAdmin();
    isUsedForPartitionedRegionBucket = internalRegionArgs.isUsedForPartitionedRegionBucket();
    isUsedForMetaRegion = internalRegionArgs.isUsedForMetaRegion();
    isMetaRegionWithTransactions = internalRegionArgs.isMetaRegionWithTransactions();
    isUsedForSerialGatewaySenderQueue = internalRegionArgs.isUsedForSerialGatewaySenderQueue();
    isUsedForParallelGatewaySenderQueue =
        internalRegionArgs.isUsedForParallelGatewaySenderQueue();
    serialGatewaySender = internalRegionArgs.getSerialGatewaySender();
    if (internalRegionArgs.getCacheServiceProfiles() != null) {
      addCacheServiceProfiles(internalRegionArgs);
    }

    if (!isUsedForMetaRegion && !isUsedForPartitionedRegionAdmin
        && !isUsedForPartitionedRegionBucket && !isUsedForSerialGatewaySenderQueue
        && !isUsedForParallelGatewaySenderQueue) {
      filterProfile = new FilterProfile(this);
    }

    // initialize client to server proxy
    serverRegionProxy =
        getPoolName() != null ? serverRegionProxyConstructor.create(this) : null;
    imageState = new UnsharedImageState(getPoolName() != null,
        getDataPolicy().withReplication() || getDataPolicy().isPreloaded(),
        getAttributes().getDataPolicy().withPersistence(), stopper);

    // prevent internal regions from participating in a TX
    supportsTX = !isSecret() && !isUsedForPartitionedRegionAdmin() && !isUsedForMetaRegion()
        || isMetaRegionWithTransactions();

    testCallable = internalRegionArgs.getTestCallable();
    eventTracker = createEventTracker();

    versionVector = createRegionVersionVector();
  }

  private void addCacheServiceProfiles(InternalRegionArguments internalRegionArgs) {
    cacheServiceProfileUpdateLock.lock();
    try {
      cacheServiceProfiles.putAll(internalRegionArgs.getCacheServiceProfiles());
    } finally {
      cacheServiceProfileUpdateLock.unlock();
    }
  }

  protected EventTracker createEventTracker() {
    return NonDistributedEventTracker.getInstance();
  }

  private RegionMap createRegionMap(InternalRegionArguments internalRegionArgs) {
    RegionMap result = null;
    if (diskRegion != null) {
      result = diskRegion.useExistingRegionMap(this);
    }
    if (result == null) {
      RegionMap.Attributes ma = new RegionMap.Attributes();
      ma.statisticsEnabled = statisticsEnabled;
      ma.loadFactor = loadFactor;
      ma.initialCapacity = initialCapacity;
      ma.concurrencyLevel = concurrencyLevel;
      result = regionMapConstructor.create(this, ma, internalRegionArgs);
    }
    return result;
  }

  /**
   * Other region classes may track events using different mechanisms than EventTrackers or may not
   * track events at all
   */
  public EventTracker getEventTracker() {
    return eventTracker;
  }

  /**
   * returns the regions version-vector
   */
  @Override
  public RegionVersionVector getVersionVector() {
    return versionVector;
  }

  /**
   * returns object used to guard the size() operation during tombstone removal
   */
  Object getSizeGuard() {
    if (!getConcurrencyChecksEnabled()) {
      return new Object();
    }
    return fullPath; // avoids creating another sync object - could be anything unique to
    // this region
  }

  protected RegionVersionVector createRegionVersionVector() {
    if (getConcurrencyChecksEnabled()) {
      return createVersionVector();
    }
    return null;
  }

  /**
   * initializes a new version vector for this region
   */
  private RegionVersionVector createVersionVector() {
    RegionVersionVector regionVersionVector = RegionVersionVector.create(getVersionMember(), this);

    if (getDataPolicy().withPersistence()) {
      // copy the versions that we have recovered from disk into
      // the version vector.
      RegionVersionVector diskVector = diskRegion.getRegionVersionVector();
      regionVersionVector.recordVersions(diskVector.getCloneForTransmission());
    } else if (!getDataPolicy().withStorage()) {
      // version vectors are currently only necessary in empty regions for
      // tracking canonical member IDs
      regionVersionVector.turnOffRecordingForEmptyRegion();
    }
    if (serverRegionProxy != null) {
      regionVersionVector.setIsClientVector();
    }
    cache.getDistributionManager().addMembershipListener(regionVersionVector);
    return regionVersionVector;
  }

  @Override
  protected void updateEntryExpiryPossible() {
    super.updateEntryExpiryPossible();
    if (!isEntryExpiryPossible()) {
      // since expiration is no longer possible cleanup the tasks
      cancelAllEntryExpiryTasks();
    }
  }

  boolean isCacheClosing() {
    return cache.isClosed();
  }

  @Override
  public RegionEntry getRegionEntry(Object key) {
    return entries.getEntry(key);
  }

  /**
   * Test hook - returns the version stamp for an entry in the form of a version tag
   *
   * @return the entry version information
   */
  @Override
  public VersionTag getVersionTag(Object key) {
    Region.Entry entry = getEntry(key, true);
    VersionTag tag = null;
    if (entry instanceof EntrySnapshot) {
      tag = ((EntrySnapshot) entry).getVersionTag();
    } else if (entry instanceof NonTXEntry) {
      tag = ((NonTXEntry) entry).getRegionEntry().getVersionStamp().asVersionTag();
    }
    return tag;
  }

  /**
   * removes any destroyed entries from the region and clear the destroyedKeys assert: Caller must
   * be holding writeLock on is
   */
  private void destroyEntriesAndClearDestroyedKeysSet() {
    ImageState imageState = getImageState();
    Iterator iterator = imageState.getDestroyedEntries();
    while (iterator.hasNext()) {
      Object key = iterator.next();
      // destroy the entry which has value Token.DESTROYED
      // If it is Token.DESTROYED then only destroy it.
      entries.removeIfDestroyed(key);
    }
  }

  /**
   * @since GemFire 5.7
   */
  private final ServerRegionProxy serverRegionProxy;

  private final InternalDataView sharedDataView;

  @Override
  public ServerRegionProxy getServerProxy() {
    return serverRegionProxy;
  }

  @Override
  public boolean hasServerProxy() {
    return serverRegionProxy != null;
  }

  /**
   * Returns true if the ExpiryTask is currently allowed to expire.
   */
  protected boolean isExpirationAllowed(ExpiryTask expiry) {
