/*
 * 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.cache.asyncqueue.internal.AsyncEventQueueImpl.getAsyncEventQueueIdFromSenderId;
import static org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl.getSenderIdFromAsyncEventQueueId;

import java.io.IOException;
import java.io.InputStream;
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.NoSuchElementException;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.SystemFailure;
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.CacheClosedException;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.LossAction;
import org.apache.geode.cache.MembershipAttributes;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionAccessException;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionDistributionException;
import org.apache.geode.cache.RegionMembershipListener;
import org.apache.geode.cache.ResumptionAction;
import org.apache.geode.cache.RoleException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.execute.ResultSender;
import org.apache.geode.cache.persistence.PersistentReplicatesOfflineException;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.LockServiceDestroyedException;
import org.apache.geode.distributed.Role;
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.DistributionAdvisor.ProfileVisitor;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.ReplyProcessor21;
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.internal.Assert;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.InitialImageOperation.GIIStatus;
import org.apache.geode.internal.cache.RegionMap.ARMLockTestHook;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.event.DistributedEventTracker;
import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.eviction.EvictableEntry;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultWaiter;
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.execute.metrics.FunctionStats;
import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager;
import org.apache.geode.internal.cache.persistence.CreatePersistentRegionProcessor;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisor;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisorImpl;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
import org.apache.geode.internal.cache.persistence.PersistentMemberView;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.tx.RemoteClearMessage;
import org.apache.geode.internal.cache.tx.RemoteDestroyMessage;
import org.apache.geode.internal.cache.tx.RemoteFetchVersionMessage;
import org.apache.geode.internal.cache.tx.RemoteFetchVersionMessage.FetchVersionResponse;
import org.apache.geode.internal.cache.tx.RemoteInvalidateMessage;
import org.apache.geode.internal.cache.tx.RemotePutMessage;
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.VersionTag;
import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException;
import org.apache.geode.internal.cache.wan.GatewaySenderConfigurationException;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.sequencelog.RegionLogger;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

@SuppressWarnings("deprecation")
public class DistributedRegion extends LocalRegion implements InternalDistributedRegion {
  private static final Logger logger = LogService.getLogger();

  /** causes cache profile to be added to afterRemoteRegionCreate notification for testing */
  @MutableForTesting
  public static boolean TEST_HOOK_ADD_PROFILE = false;

  /** Used to sync accesses to this.dlockService to allow lazy construction */
  private final Object dlockMonitor = new Object();

  final CacheDistributionAdvisor distAdvisor;
  private final SenderIdMonitor senderIdMonitor;

  /**
   * GuardedBy {@link #dlockMonitor}
   */
  private DistributedLockService dlockService;

  final AdvisorListener advisorListener = new AdvisorListener();

  /** Set of currently missing required roles */
  private final HashSet<Role> missingRequiredRoles = new HashSet<>();

  /** True if this region is currently missing any required roles */
  private volatile boolean isMissingRequiredRoles = false;

  /**
   * True if this region is has any required roles defined and the LossAction is either NO_ACCESS or
   * LIMITED_ACCESS. Reliability checks will only happen if this is true.
   */
  private final boolean requiresReliabilityCheck;

  /**
   * Latch that is opened after initialization waits for required roles up to the
   * <a href="DistributedSystem#member-timeout">member-timeout </a>.
   */
  private final StoppableCountDownLatch initializationLatchAfterMemberTimeout;

  private final PersistenceAdvisor persistenceAdvisor;

  private final PersistentMemberID persistentId;

  /**
   * This boolean is set to false when this region is non-persistent, but there are persistent
   * members in the distributed system to which all region modifications should be forwarded see bug
   * 45186
   */
  private volatile boolean generateVersionTag = true;

  /** Tests can set this to true and ignore reliability triggered reconnects */
  @MutableForTesting
  public static boolean ignoreReconnect = false;

  /**
   * Lock to prevent multiple threads on this member from performing a clear at the same time.
   */
  private final Object clearLock = new Object();
  private final ReentrantReadWriteLock failedInitialImageLock = new ReentrantReadWriteLock(true);

  @MakeNotStatic
  private static final AtomicBoolean loggedNetworkPartitionWarning = new AtomicBoolean(false);

  /** Creates a new instance of DistributedRegion */
  protected DistributedRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
      InternalCache cache, InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock) {
    super(regionName, attrs, parentRegion, cache, internalRegionArgs, statisticsClock);
    initializationLatchAfterMemberTimeout =
        new StoppableCountDownLatch(getCancelCriterion(), 1);
    distAdvisor = createDistributionAdvisor(internalRegionArgs);
    senderIdMonitor = createSenderIdMonitor();

    if (getDistributionManager().getConfig().getEnableNetworkPartitionDetection()
        && !isInternalRegion() && !attrs.getScope().isAck() && !doesNotDistribute()
        && attrs.getDataPolicy().withStorage()) {
      logger.warn(
          "Region {} is being created with scope {} but enable-network-partition-detection is enabled in the distributed system.  This can lead to cache inconsistencies if there is a network failure.",
          new Object[] {regionName, attrs.getScope()});
    }
    if (!getDistributionManager().getConfig().getEnableNetworkPartitionDetection()
        && attrs.getDataPolicy().withPersistence()
        && !loggedNetworkPartitionWarning.getAndSet(true)) {
      logger.warn(
          "Creating persistent region {}, but enable-network-partition-detection is set to false. Running with network partition detection disabled can lead to an unrecoverable system in the event of a network split.",
          regionName);
    }

    boolean setRequiresReliabilityCheck = attrs.getMembershipAttributes().hasRequiredRoles() &&
    // note that the following includes NO_ACCESS, LIMITED_ACCESS,
        !attrs.getMembershipAttributes().getLossAction().isAllAccess()
        && !attrs.getMembershipAttributes().getLossAction().isReconnect();

    // this optimization is safe for as long as Roles and Required Roles are
    // immutable
    // if this VM fulfills all required roles, make requiresReliabilityCheck
    // false
    Set<Role> reqRoles = new HashSet<>(attrs.getMembershipAttributes().getRequiredRoles());
    reqRoles.removeAll(getSystem().getDistributedMember().getRoles());
    if (reqRoles.isEmpty()) {
      setRequiresReliabilityCheck = false;
    }

    requiresReliabilityCheck = setRequiresReliabilityCheck;

    if (internalRegionArgs.isUsedForPartitionedRegionBucket()) {
      persistenceAdvisor = internalRegionArgs.getPersistenceAdvisor();
    } else if (allowsPersistence()) {
      // TODO: prpersist - using this lock service is a hack. Maybe? Or maybe
      // it's ok if we have one (rarely used) lock service for many operations?
      // What does the resource manager do?
      DistributedLockService dl = cache.getPartitionedRegionLockService();
      try {
        // TODO: prpersist - this is just a quick and dirty storage mechanism so that
        // I can test the storage.
        DiskRegionStats diskStats;
        PersistentMemberView storage;
        if (getDataPolicy().withPersistence()) {
          storage = getDiskRegion();
          diskStats = getDiskRegion().getStats();
        } else {
          storage = new InMemoryPersistentMemberView();
          diskStats = null;
        }
        PersistentMemberManager memberManager = cache.getPersistentMemberManager();
        persistenceAdvisor = new PersistenceAdvisorImpl(distAdvisor, dl, storage,
            getFullPath(), diskStats, memberManager);
      } catch (Exception ignore) {
        throw new InternalGemFireError("Couldn't recover persistence");
      }
    } else {
      persistenceAdvisor = null;
    }
    if (persistenceAdvisor != null) {
      persistentId = persistenceAdvisor.generatePersistentID();
    } else {
      persistentId = null;
    }
  }

  @Override
  protected EventTracker createEventTracker() {
    EventTracker tracker = new DistributedEventTracker(this);
    tracker.start();
    return tracker;
  }

  /**
   * Record the event state from image provider
   *
   * @param provider the member that provided the initial image and event state
   */
  protected void recordEventStateFromImageProvider(InternalDistributedMember provider) {
    // No Op. Only Bucket region will initiate event states
  }

  /**
   * Intended for used during construction of a DistributedRegion
   *
   * @return the advisor to be used by the region
   */
  protected CacheDistributionAdvisor createDistributionAdvisor(
      InternalRegionArguments internalRegionArgs) {
    // Warning: potential early escape of object before full construction
    return CacheDistributionAdvisor.createCacheDistributionAdvisor(this);
  }

  /**
   * Does this region support persistence?
   */
  public boolean allowsPersistence() {
    return true;
  }

  @Override
  boolean requiresOneHopForMissingEntry(EntryEventImpl event) {
    // received from another member - don't use one-hop
    if (event.isOriginRemote()) {
      return false;
    }
    // local ops aren't distributed
    if (event.getOperation().isLocal()) {
      return false;
    }
    // if it already has a valid version tag it can go out with a DistributedCacheOperation
    if (event.getVersionTag() != null && event.getVersionTag().getRegionVersion() > 0) {
      return false;
    }
    // if we're not allowed to generate a version tag we need to send it to someone who can
    if (!generateVersionTag) {
      return true;
    }
    return getConcurrencyChecksEnabled() && (getServerProxy() == null) && !isTX()
        && scope.isDistributed() && !getDataPolicy().withReplication();
  }


  @Override
  public boolean virtualPut(EntryEventImpl event, boolean ifNew, boolean ifOld,
      Object expectedOldValue, boolean requireOldValue, long lastModified,
      boolean overwriteDestroyed, boolean invokeCallbacks, boolean throwConcurrentModificaiton)
      throws TimeoutException, CacheWriterException {
    final boolean isTraceEnabled = logger.isTraceEnabled();

    Lock dlock = null;
    if (scope.isGlobal() && // lock only applies to global scope
        !event.isOriginRemote() && // only if operation originating locally
        !event.isNetSearch() && // search and load processor handles own locking
        !event.isNetLoad() &&
        // @todo darrel/kirk: what about putAll?
        !event.isLocalLoad() && !event.isSingleHopPutOp()) {
      // Single Hop Op means dlock is already taken at origin node.
      dlock = getDistributedLockIfGlobal(event.getKey());
    }
    if (isTraceEnabled) {
      logger.trace("virtualPut invoked for event {}", event);
    }
    try {
      if (!hasSeenEvent(event)) {
        if (requiresOneHopForMissingEntry(event)) {
          // bug #45704: see if a one-hop must be done for this operation
          RegionEntry re = getRegionEntry(event.getKey());
          if (re == null /* || re.isTombstone() */ || !generateVersionTag
              || getDataPolicy() == DataPolicy.NORMAL
              || getDataPolicy() == DataPolicy.PRELOADED) {
            // Let NORMAL and PRELOAD to behave the same as EMPTY
            if (!event.isBulkOpInProgress() || getDataPolicy().withStorage()) {
              // putAll will send a single one-hop for empty regions. for other missing entries
              // we need to get a valid version number before modifying the local cache
              boolean didDistribute = RemotePutMessage.distribute(event, lastModified, false, false,
                  expectedOldValue, requireOldValue, !generateVersionTag);

              if (!didDistribute && isTraceEnabled) {
                logger.trace("Unable to perform one-hop messaging");
              }
              if (!generateVersionTag && !didDistribute) {
                throw new PersistentReplicatesOfflineException();
              }
              if (didDistribute) {
                if (isTraceEnabled) {
                  logger.trace("Event after remotePut operation: {}", event);
                }
                if (event.getVersionTag() == null) {
                  // if the event wasn't applied by the one-hop replicate it will not have a version
                  // tag
                  // and so should not be applied to this cache
                  return false;
                }
              }
            }
          }
        }
        return super.virtualPut(event, ifNew, ifOld, expectedOldValue, requireOldValue,
            lastModified, overwriteDestroyed, invokeCallbacks, throwConcurrentModificaiton);
      } else {
        if (event.getDeltaBytes() != null && event.getRawNewValue() == null) {
          // This means that this event has delta bytes but no full value.
          // Request the full value of this event.
          // The value in this vm may not be same as this event's value.
          throw new InvalidDeltaException(
              "Cache encountered replay of event containing delta bytes for key " + event.getKey());
        }
        // if the listeners have already seen this event, then it has already
        // been successfully applied to the cache. Distributed messages and
        // return
        if (logger.isDebugEnabled()) {
          logger.debug("DR.virtualPut: this cache has already seen this event {}", event);
        }

        // Fix 39014: when hasSeenEvent, put will still distribute
        // event, but putAll did not. We add the logic back here, not to put
        // back into DR.distributeUpdate() because we moved this part up into
        // LR.basicPutPart3 in purpose.
        if (event.isBulkOpInProgress() && !event.isOriginRemote()) {
          event.getPutAllOperation().addEntry(event, true);
        }

        /*
         * doing this so that other VMs will apply this no matter what. If it is an "update" they
         * will not apply it if they don't have the key. Because this is probably a retry, it will
         * never get applied to this local AbstractRegionMap, and so will never be flipped to a
         * 'create'
         */
        event.makeCreate();
        if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
          distributeUpdate(event, lastModified, ifNew, ifOld, expectedOldValue, requireOldValue);
          event.invokeCallbacks(this, true, true);
        }
        return true;
      }
    } finally {
      if (dlock != null) {
        dlock.unlock();
      }
    }
  }

  @Override
  RegionEntry basicPutEntry(EntryEventImpl event, long lastModified)
      throws TimeoutException, CacheWriterException {

    final boolean isTraceEnabled = logger.isTraceEnabled();

    if (isTraceEnabled) {
      logger.trace("basicPutEntry invoked for event {}", event);
    }
    if (requiresOneHopForMissingEntry(event)) {
      // bug #45704: see if a one-hop must be done for this operation
      RegionEntry re = getRegionEntry(event.getKey());
      if (re == null /* || re.isTombstone() */ || !generateVersionTag) {
        final boolean ifNew = false;
        final boolean ifOld = false;
        boolean didDistribute = RemotePutMessage.distribute(event, lastModified, ifNew, ifOld, null,
            false, !generateVersionTag);
        if (!generateVersionTag && !didDistribute) {
          throw new PersistentReplicatesOfflineException();
        }
        if (didDistribute && isTraceEnabled) {
          logger.trace("Event after remotePut for basicPutEntry: {}", event);
        }
      }
    }
    return super.basicPutEntry(event, lastModified);
  }

  @Override
  void performPutAllEntry(EntryEventImpl event) {
    /*
     * force shared data view so that we just do the virtual op, accruing things in the put all
     * operation for later
     */
    if (isTX()) {
      event.getPutAllOperation().addEntry(event);
    } else {
      getSharedDataView().putEntry(event, false, false, null, false, 0L, false);
    }
  }

  @Override
  void performRemoveAllEntry(EntryEventImpl event) {
    // force shared data view so that we just do the virtual op, accruing things in the bulk
    // operation for later
    if (isTX()) {
      event.getRemoveAllOperation().addEntry(event);
    } else {
      basicDestroy(event, true, null);
    }
  }

  /**
   * distribution and listener notification
   */
  @Override
  public void basicPutPart3(EntryEventImpl event, RegionEntry entry, boolean isInitialized,
      long lastModified, boolean invokeCallbacks, boolean ifNew, boolean ifOld,
      Object expectedOldValue, boolean requireOldValue) {

    distributeUpdate(event, lastModified, false, false, null, false);
    super.basicPutPart3(event, entry, isInitialized, lastModified, invokeCallbacks, ifNew, ifOld,
        expectedOldValue, requireOldValue);
  }

  /** distribute an update operation */
  protected void distributeUpdate(EntryEventImpl event, long lastModified, boolean ifNew,
      boolean ifOld, Object expectedOldValue, boolean requireOldValue) {
    // an update from a netSearch is not distributed
    if (!event.isOriginRemote() && !event.isNetSearch() && !event.isBulkOpInProgress()) {
      boolean distribute = !event.getInhibitDistribution();
      // this has already been distributed by a one-hop operation
      if (distribute) {
        // before distribute: DR's put, it has notified gateway sender earlier
        UpdateOperation op = new UpdateOperation(event, lastModified);
        if (logger.isTraceEnabled()) {
          logger.trace("distributing operation for event : {} : for region : {}", event,
              getName());
        }
        op.distribute();
      }
    }
  }

  @Override
  public boolean hasSeenEvent(EntryEventImpl event) {
    boolean isDuplicate;

    isDuplicate = getEventTracker().hasSeenEvent(event);
    if (isDuplicate) {
      markEventAsDuplicate(event);
      if (event.getVersionTag() == null && event.getRegion().getConcurrencyChecksEnabled()
          && event.getEventId() != null) {
        // For a retry, versionTag can be missing possibly due to no versionTag recorded during gii.
        findAndSetVersionTag(event);
        if (event.getVersionTag() == null) {
          logger.info("No version tag can be found in cluster when retrying the following event {}",
              event);
        }
      }
    } else {
      // bug #48205 - a retried PR operation may already have a version assigned to it
      // in another VM
      if (event.isPossibleDuplicate() && event.getRegion().getConcurrencyChecksEnabled()
          && event.getVersionTag() == null && event.getEventId() != null) {
        findAndSetVersionTag(event);
      }
    }
    return isDuplicate;
  }

  @VisibleForTesting
  void findAndSetVersionTag(EntryEventImpl event) {
    boolean isBulkOp = event.getOperation().isPutAll() || event.getOperation().isRemoveAll();
    VersionTag tag =
        FindVersionTagOperation.findVersionTag(event.getRegion(), event.getEventId(), isBulkOp);
    event.setVersionTag(tag);
  }

  @VisibleForTesting
  void markEventAsDuplicate(EntryEventImpl event) {
    event.setPossibleDuplicate(true);
    if (getConcurrencyChecksEnabled() && event.getVersionTag() == null) {
      if (event.isBulkOpInProgress()) {
        event.setVersionTag(getEventTracker().findVersionTagForBulkOp(event.getEventId()));
      } else {
        event.setVersionTag(getEventTracker().findVersionTagForSequence(event.getEventId()));
      }
    }
  }

  void setGeneratedVersionTag(boolean generateVersionTag) {
    // there is at-least one other persistent member, so turn on concurrencyChecks
    enableConcurrencyChecks();

    this.generateVersionTag = generateVersionTag;
  }

  protected boolean getGenerateVersionTag() {
    return generateVersionTag;
  }

  @Override
  boolean shouldGenerateVersionTag(RegionEntry entry, EntryEventImpl event) {
    if (logger.isTraceEnabled()) {
      logger.trace(
          "shouldGenerateVersionTag this.generateVersionTag={} ccenabled={} dataPolicy={} event:{}",
          generateVersionTag, getConcurrencyChecksEnabled(), getDataPolicy(), event);
    }
    if (!getConcurrencyChecksEnabled() || getDataPolicy() == DataPolicy.EMPTY
        || !generateVersionTag) {
      return false;
    }
    if (getServerProxy() != null) { // client
      return false;
    }
    if (event.getVersionTag() != null && !event.getVersionTag().isGatewayTag()) {
      if (logger.isDebugEnabled()) {
        logger.debug("Not to create a new version tag for retried event {}", event);
      }
      return false;
    }
    if (event.getOperation().isLocal()) { // bug #45402 - localDestroy generated a version tag
      return false;
    }
    if (!event.isOriginRemote() && getDataPolicy().withReplication()) {
      return true;
    }
    if (!getDataPolicy().withReplication() && !getDataPolicy().withPersistence()) {
      // do not generate a version stamp in a region that has no replication if it's not based
      // on an existing version from a replicate region
      return entry.getVersionStamp().hasValidVersion();
    }
    if (!event.isOriginRemote() && event.getDistributedMember() != null) {
      if (!event.getDistributedMember().equals(getMyId())) {
        return event.getVersionTag() == null; // one-hop remote message
      }
    }
    return false;
  }

  /**
   * Throws RegionAccessException if required roles are missing and the LossAction is NO_ACCESS
   *
   * @throws RegionAccessException if required roles are missing and the LossAction is NO_ACCESS
   */
  @Override
  protected void checkForNoAccess() {
    if (requiresReliabilityCheck && isMissingRequiredRoles) {
      if (getMembershipAttributes().getLossAction().isNoAccess()) {
        synchronized (missingRequiredRoles) {
          if (!isMissingRequiredRoles) {
            return;
          }
          Set<Role> roles = Collections.unmodifiableSet(new HashSet<>(missingRequiredRoles));
          throw new RegionAccessException(
              String.format(
                  "Operation is disallowed by LossAction %s because these required roles are missing: %s.",
                  getMembershipAttributes().getLossAction(), roles),
              getFullPath(), roles);
        }
      }
    }
  }

  /**
   * Throws RegionAccessException is required roles are missing and the LossAction is either
   * NO_ACCESS or LIMITED_ACCESS.
   *
   * @throws RegionAccessException if required roles are missing and the LossAction is either
   *         NO_ACCESS or LIMITED_ACCESS
   */
  @Override
  public void checkForLimitedOrNoAccess() {
    if (requiresReliabilityCheck && isMissingRequiredRoles) {
      if (getMembershipAttributes().getLossAction().isNoAccess()
          || getMembershipAttributes().getLossAction().isLimitedAccess()) {
        synchronized (missingRequiredRoles) {
          if (!isMissingRequiredRoles) {
            return;
          }
          Set<Role> roles = Collections.unmodifiableSet(new HashSet<>(missingRequiredRoles));
          Assert.assertTrue(!roles.isEmpty());
          throw new RegionAccessException(
              String.format(
                  "Operation is disallowed by LossAction %s because these required roles are missing: %s.",
                  getMembershipAttributes().getLossAction(), roles),
              getFullPath(), roles);
        }
      }
    }
  }

  @Override
  public void handleReliableDistribution(Set successfulRecipients) {
    handleReliableDistribution(successfulRecipients, Collections.emptySet(),
        Collections.emptySet());
  }

  private void handleReliableDistribution(Set successfulRecipients, Set otherRecipients1,
      Set otherRecipients2) {
    if (requiresReliabilityCheck) {
      MembershipAttributes ra = getMembershipAttributes();
      // determine the successful roles
      Set<Role> roles = new HashSet<>();
      for (Object successfulRecipient : successfulRecipients) {
        InternalDistributedMember mbr = (InternalDistributedMember) successfulRecipient;
        if (mbr != null) {
          roles.addAll(mbr.getRoles());
        }
      }
      for (Object anOtherRecipients1 : otherRecipients1) {
        InternalDistributedMember mbr = (InternalDistributedMember) anOtherRecipients1;
        if (mbr != null) {
          roles.addAll(mbr.getRoles());
        }
      }
      for (Object anOtherRecipients2 : otherRecipients2) {
        InternalDistributedMember mbr = (InternalDistributedMember) anOtherRecipients2;
        if (mbr != null) {
          roles.addAll(mbr.getRoles());
        }
      }
      // determine the missing roles
      Set<Role> failedRoles = new HashSet<>(ra.getRequiredRoles());
      failedRoles.removeAll(roles);
      if (failedRoles.isEmpty()) {
        return;
      }

      throw new RegionDistributionException(
          String.format("Operation distribution may have failed to notify these required roles: %s",
              failedRoles),
          getFullPath(), failedRoles);
    }
  }

  /**
   * Called when we do a distributed operation and don't have anyone to distributed it too. Since
   * this is only called when no distribution was done (i.e. no recipients) we do not check
   * isMissingRequiredRoles because it might not longer be true due to race conditions
   *
   * @return false if this region has at least one required role and queuing is configured. Returns
   *         true if sending to no one is ok.
   * @throws RoleException if a required role is missing and the LossAction is either NO_ACCESS or
   *         LIMITED_ACCESS.
   * @since GemFire 5.0
   */
  boolean isNoDistributionOk() {
    if (requiresReliabilityCheck) {
      MembershipAttributes ra = getMembershipAttributes();
      Set<Role> failedRoles = ra.getRequiredRoles();
      throw new RegionDistributionException(
          String.format("Operation distribution was not done to these required roles: %s",
              failedRoles),
          getFullPath(), failedRoles);
    }
    return true;
  }

  /**
   * returns true if this Region does not distribute its operations to other members.
   *
   * @since GemFire 6.0
   * @see HARegion#localDestroyNoCallbacks(Object)
   */
  public boolean doesNotDistribute() {
    return false;
  }

  @Override
  public boolean shouldSyncForCrashedMember(InternalDistributedMember id) {
    return !doesNotDistribute() && super.shouldSyncForCrashedMember(id);
  }

  @Override
  public boolean requiresReliabilityCheck() {
    return requiresReliabilityCheck;
  }

  /**
   * Returns true if the ExpiryTask is currently allowed to expire.
   * <p>
   * If the region is in NO_ACCESS due to reliability configuration, then no expiration actions are
   * allowed.
   * <p>
