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

import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.DynamicRegionFactory;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.Scope;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionAdvisee;
import org.apache.geode.distributed.internal.DistributionAdvisor;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MessageWithReply;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.InitialImageAdvice;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.event.EventSequenceNumberHolder;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor.PartitionProfile;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.logging.LogService;

/**
 * This message processor handles creation and initial exchange of
 * org.apache.geode.internal.cache.CacheDistributionAdvisor.Profiles. It represents creation of a
 * {@link CacheDistributionAdvisee}. Name remains CreateRegion to avoid merge conflicts.
 */
public class CreateRegionProcessor implements ProfileExchangeProcessor {
  private static final Logger logger = LogService.getLogger();

  protected CacheDistributionAdvisee newRegion;

  /** Creates a new instance of CreateRegionProcessor */
  public CreateRegionProcessor(CacheDistributionAdvisee newRegion) {
    this.newRegion = newRegion;
  }

  /** this method tells other members that the region is being created */
  @Override
  public void initializeRegion() {
    InternalDistributedSystem system = this.newRegion.getSystem();
    // try 5 times, see CreateRegionMessage#skipDuringInitialization
    for (int retry = 0; retry < 5; retry++) {
      Set recps = getRecipients();

      if (logger.isDebugEnabled()) {
        logger.debug("Creating region {}", this.newRegion);
      }

      if (recps.isEmpty()) {
        if (logger.isDebugEnabled()) {
          logger.debug("CreateRegionProcessor.initializeRegion, no recipients, msg not sent");
        }
        this.newRegion.getDistributionAdvisor().setInitialized();

        ((LocalRegion) this.newRegion).getEventTracker().setInitialized();
        return;
      }

      CreateRegionReplyProcessor replyProc = new CreateRegionReplyProcessor(recps);
      newRegion.registerCreateRegionReplyProcessor(replyProc);

      boolean useMcast = false; // multicast is disabled for this message for now
      CreateRegionMessage msg = getCreateRegionMessage(recps, replyProc, useMcast);

      // since PR buckets can be created during cache entry operations, enable
      // severe alert processing if we're creating one of them
      if (((LocalRegion) newRegion).isUsedForPartitionedRegionBucket()) {
        replyProc.enableSevereAlertProcessing();
        msg.severeAlertCompatible = true;
      }

      this.newRegion.getDistributionManager().putOutgoing(msg);
      // this was in a while() loop, which is incorrect use of a reply processor.
      // Reply procs are deregistered when they return from waitForReplies
      try {
        // Don't allow a region to be created if the distributed system is
        // disconnecting
        this.newRegion.getCache().getCancelCriterion().checkCancelInProgress(null);

        // This isn't right. We should disable region creation in general, not just
        // the remote case here...
        // // Similarly, don't allow new regions to be created if the cache is closing
        try {
          replyProc.waitForRepliesUninterruptibly();
          if (!replyProc.needRetry()) {
            break;
          }
        } catch (ReplyException e) {
          Throwable t = e.getCause();
          if (t instanceof IllegalStateException) {
            // region is incompatible with region in another cache
            throw (IllegalStateException) t;
          }
          e.handleCause();
          break;
        }
      } finally {
        replyProc.cleanup();
        ((LocalRegion) this.newRegion).getEventTracker().setInitialized();
        if (((LocalRegion) this.newRegion).isUsedForPartitionedRegionBucket()) {
          if (logger.isDebugEnabled()) {
            logger.debug("initialized bucket event tracker: {}",
                ((LocalRegion) this.newRegion).getEventTracker());
          }
        }
      }
    } // while
    // tell advisor that it has been initialized since a profile exchange occurred
    this.newRegion.getDistributionAdvisor().setInitialized();
  }

  protected Set getRecipients() {
    DistributionAdvisee parent = this.newRegion.getParentAdvisee();
    Set recps = null;
    if (parent == null) { // root region, all recipients
      InternalDistributedSystem system = this.newRegion.getSystem();
      recps = system.getDistributionManager().getOtherDistributionManagerIds();
    } else {
      // get recipients that have the parent region defined as distributed.
      recps = getAdvice();
    }
    return recps;
  }

  @Override
  public InitialImageAdvice getInitialImageAdvice(InitialImageAdvice previousAdvice) {
    return newRegion.getCacheDistributionAdvisor().adviseInitialImage(previousAdvice);
  }

  private Set getAdvice() {
    if (this.newRegion instanceof BucketRegion) {
      return ((Bucket) this.newRegion).getBucketAdvisor().adviseProfileExchange();
    } else {
      DistributionAdvisee rgn = this.newRegion.getParentAdvisee();
      DistributionAdvisor advisor = rgn.getDistributionAdvisor();
      return advisor.adviseGeneric();
    }
  }

  protected CreateRegionMessage getCreateRegionMessage(Set recps, ReplyProcessor21 proc,
      boolean useMcast) {
    CreateRegionMessage msg = new CreateRegionMessage();
    msg.regionPath = this.newRegion.getFullPath();
    msg.profile = (CacheProfile) this.newRegion.getProfile();
    msg.processorId = proc.getProcessorId();
    msg.concurrencyChecksEnabled = this.newRegion.getAttributes().getConcurrencyChecksEnabled();
    msg.setMulticast(useMcast);
    msg.setRecipients(recps);
    return msg;
  }

  @Override
  public void setOnline(InternalDistributedMember target) {
    // nothing
  }

  class CreateRegionReplyProcessor extends ReplyProcessor21 {

    CreateRegionReplyProcessor(Set members) {
      super((InternalDistributedSystem) CreateRegionProcessor.this.newRegion.getCache()
          .getDistributedSystem(), members);
    }

    private final Map<DistributedMember, Map<ThreadIdentifier, EventSequenceNumberHolder>> remoteEventStates =
        new ConcurrentHashMap<>();

    private boolean allMembersSkippedChecks = true;

    public Map<ThreadIdentifier, EventSequenceNumberHolder> getEventState(
        InternalDistributedMember provider) {
      return this.remoteEventStates.get(provider);
    }

    /**
     * true if all members skipped CreateRegionMessage#checkCompatibility(), in which case
     * CreateRegionMessage should be retried.
     */
    public boolean needRetry() {
      return this.allMembersSkippedChecks;
    }

    @SuppressWarnings("unchecked")
    @Override
    public void process(DistributionMessage msg) {
      Assert.assertTrue(msg instanceof CreateRegionReplyMessage,
          "CreateRegionProcessor is unable to process message of type " + msg.getClass());
      CreateRegionReplyMessage reply = (CreateRegionReplyMessage) msg;
      LocalRegion lr = (LocalRegion) newRegion;
      if (logger.isDebugEnabled()) {
        logger.debug("CreateRegionProcessor processing {}", msg);
      }
      try {
        if (reply.profile != null) {
          if (newRegion instanceof DistributedRegion) {
            DistributedRegion dr = (DistributedRegion) newRegion;
            if (!dr.getDataPolicy().withPersistence() && reply.profile.isPersistent) {
              dr.setGeneratedVersionTag(false);
            }
          }
          if (CreateRegionMessage.isLocalAccessor(newRegion) && reply.profile.isPersistent) {
            lr.enableConcurrencyChecks();
          }

          CacheDistributionAdvisor cda =
              CreateRegionProcessor.this.newRegion.getCacheDistributionAdvisor();
          cda.putProfile(reply.profile);
          if (reply.bucketProfiles != null) {
            RegionAdvisor ra = (RegionAdvisor) cda;
            ra.putBucketRegionProfiles(reply.bucketProfiles);
          }

          // Save all event states, need to initiate the event tracker from the GII provider
          if (reply.eventState != null) {
            remoteEventStates.put(reply.getSender(),
                (Map<ThreadIdentifier, EventSequenceNumberHolder>) reply.eventState);
          }

          if (lr.isUsedForPartitionedRegionBucket()) {
            ((BucketRegion) lr).updateEventSeqNum(reply.seqKeyForWan);
          }
          // Process any delta filter-profile messages received during profile
          // exchange.
          // The pending messages are queued in the local profile.
          FilterProfile remoteFP = reply.profile.filterProfile;
          if (remoteFP != null) {
            FilterProfile localFP = ((LocalRegion) newRegion).filterProfile;
            // localFP can be null and remoteFP not null when upgrading from 7.0.1.14 to 7.0.1.15
            if (localFP != null) {
              List messages = localFP.getQueuedFilterProfileMsgs(reply.getSender());
              // Thread init level is set since region is used during CQ registration.
              final InitializationLevel oldLevel =
                  LocalRegion.setThreadInitLevelRequirement(ANY_INIT);
              try {
                remoteFP.processQueuedFilterProfileMsgs(messages);
              } finally {
                LocalRegion.setThreadInitLevelRequirement(oldLevel);
                localFP.removeQueuedFilterProfileMsgs(reply.getSender());
              }
            }
          }
        }
        if (reply.destroyedId != null && newRegion instanceof DistributedRegion) {
          DistributedRegion dr = (DistributedRegion) newRegion;
          dr.getPersistenceAdvisor().removeMember(reply.destroyedId);
        }
        if (!reply.skippedCompatibilityChecks) {
          allMembersSkippedChecks = false;
        }
      } finally {
        // invoke super.process() even in case of exceptions (bug #41556)
        if (logger.isDebugEnabled()) {
          logger.debug("CreateRegionProcessor invoking super.process()");
        }
        super.process(msg);
      }
    }

    /**
     * IllegalStateException is an anticipated reply exception. Receiving multiple replies with this
     * exception is normal.
     */
    @Override
    protected boolean logMultipleExceptions() {
      return false;
    }
  }

  public static class CreateRegionMessage extends HighPriorityDistributionMessage
      implements MessageWithReply {

    public boolean concurrencyChecksEnabled;
    protected String regionPath;
    protected CacheProfile profile;
    protected int processorId;

    private transient boolean incompatible = false;
    private transient ReplyException replyException;
    private transient CacheProfile replyProfile;
    private transient ArrayList replyBucketProfiles;
    private transient Object eventState;
    protected transient boolean severeAlertCompatible;
    private transient boolean skippedCompatibilityChecks;

    @Override
    public int getProcessorId() {
      return this.processorId;
    }

    @Override
    public boolean isSevereAlertCompatible() {
      return severeAlertCompatible;
    }

    @Override
    public boolean sendViaUDP() {
      return true;
    }

    @Override
    protected void process(ClusterDistributionManager dm) {
      // Set thread local flag to allow entrance through initialization Latch
      final InitializationLevel oldLevel = LocalRegion.setThreadInitLevelRequirement(ANY_INIT);
      LocalRegion lclRgn = null;

      PersistentMemberID destroyedId = null;
      try {
        // get the region from the path, but do NOT wait on initialization,
        // otherwise we could have a distributed deadlock

        InternalCache cache = dm.getExistingCache();

        // Fix for bug 42051 - Discover any regions that are in the process
        // of being destroyed
        DistributedRegion destroyingRegion = cache.getRegionInDestroy(this.regionPath);
        if (destroyingRegion != null) {
          destroyedId = destroyingRegion.getPersistentID();
        }

        lclRgn = (LocalRegion) cache.getRegion(this.regionPath);

        if (lclRgn instanceof CacheDistributionAdvisee) {
          // bug 37604 - don't return a profile if this is a bucket and the owner
          // has been locally destroyed
          if (lclRgn.isUsedForPartitionedRegionBucket()) {
            if (!((BucketRegion) lclRgn).isPartitionedRegionOpen()) {
              if (logger.isDebugEnabled()) {
                logger.debug("<Partitioned Region Closed or Locally Destroyed> {}", this);
              }
              return;
            }
          }
          handleCacheDistributionAdvisee((CacheDistributionAdvisee) lclRgn, true);
        } else {
          if (lclRgn == null) {
            // check to see if a ProxyBucketRegion (not a true region) exists
            handleCacheDistributionAdvisee(
                PartitionedRegionHelper.getProxyBucketRegion(cache, this.regionPath, false), false);
          } else {
            if (logger.isDebugEnabled()) {
              logger.debug("<lclRgn scope is not distributed. Scope={}> {}",
                  lclRgn.getAttributes().getScope(), this);
            }
          }
        }
      } catch (PRLocallyDestroyedException ignore) {
        if (logger.isDebugEnabled()) {
          logger.debug("<Region Locally Destroyed> {}", this);
        }
      } catch (RegionDestroyedException ignore) {
        if (logger.isDebugEnabled()) {
          logger.debug("<RegionDestroyed> {}", this);
        }
      } catch (CancelException ignore) {
        if (logger.isDebugEnabled()) {
          logger.debug("<CancelException> {}", this);
        }
      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable t) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        if (replyException == null) {
          replyException = new ReplyException(t);
        } else {
          logger.warn(String.format("More than one exception thrown in %s", this),
              t);
        }
      } finally {
        LocalRegion.setThreadInitLevelRequirement(oldLevel);
        CreateRegionReplyMessage replyMsg = new CreateRegionReplyMessage();
        replyMsg.profile = replyProfile;
        replyMsg.bucketProfiles = replyBucketProfiles;
        replyMsg.eventState = this.eventState;
        replyMsg.destroyedId = destroyedId;
        replyMsg.setProcessorId(this.processorId);
        replyMsg.setSender(dm.getId()); // for EventStateHelper.dataSerialize
        replyMsg.setRecipient(this.getSender());
        replyMsg.skippedCompatibilityChecks = this.skippedCompatibilityChecks;

        if (lclRgn != null && lclRgn.isUsedForPartitionedRegionBucket()) {
          replyMsg.seqKeyForWan = ((BucketRegion) lclRgn).getEventSeqNum().get();
        }
        if (replyException != null && !this.incompatible) {
          // no need to log the exception if it was caused by compatibility check
          if (logger.isDebugEnabled()) {
            logger.debug("While processing '{}', got exception, returning to sender", this,
                replyException);
          }
        }
        replyMsg.setException(replyException);
        dm.putOutgoing(replyMsg);
        if (lclRgn instanceof PartitionedRegion)
          ((PartitionedRegion) lclRgn).sendIndexCreationMsg(this.getSender());
      }
    }

    /**
     * Attempts to process this message with the specified <code>CacheDistributionAdvisee</code>.
     *
     * @param cda the CacheDistributionAdvisee to apply this profile to
     * @param isRealRegion true if CacheDistributionAdvisee is a real region
     */
    private void handleCacheDistributionAdvisee(CacheDistributionAdvisee cda,
        boolean isRealRegion) {
      if (cda == null) {
        // local region or proxy bucket region not found
        if (logger.isDebugEnabled()) {
          logger.debug("<lclRgn is null> {}", this); // matches old logging
        }
        return;
      }
      String errorMsg = null;
      if (isRealRegion) {
        // only check compatibility if this advisee is a real region
        errorMsg = checkCompatibility(cda, this.profile);
      }
      if (errorMsg != null) {
        this.incompatible = true;
        if (logger.isDebugEnabled()) {
          logger.debug("{} <replyProfile not set because errorMsg={}", this, errorMsg);
        }
        this.replyException = new ReplyException(new IllegalStateException(errorMsg));
      } else {
        if (isRealRegion) { // TODO do we need this if clause??
          // if the new member is persistent, turn on concurrency checks
          // fixes bug 45208
          if (isLocalAccessor(cda) && this.profile.isPersistent) {
            // #45934 need to set the generateVersionTag flag
            if (cda instanceof DistributedRegion) {
              DistributedRegion dr = (DistributedRegion) cda;
              if (!dr.getDataPolicy().withPersistence()) {
                dr.setGeneratedVersionTag(false);
              }
            }

            assert cda instanceof LocalRegion;
            LocalRegion lr = (LocalRegion) cda;
            lr.enableConcurrencyChecks();
          }
        }

        // #45934 don't add profile until the attributes are set correctly,
        // in particular enableConcurrencyChecks and generateVersionTag
        cda.getDistributionAdvisor().putProfile(this.profile);

        if (isRealRegion) {
          // only exchange profile if this advisee is a real region
          this.replyProfile = (CacheProfile) cda.getProfile();
          if (cda instanceof PartitionedRegion) {
            // partitioned region needs to also answer back all real bucket profiles
            PartitionedRegion pr = (PartitionedRegion) cda;
            this.replyBucketProfiles = pr.getRegionAdvisor().getBucketRegionProfiles();
          } else if (((LocalRegion) cda).isUsedForPartitionedRegionBucket()) {
            this.eventState = ((LocalRegion) cda).getEventState();
          }
        }
      }
    }

    protected String checkCompatibility(CacheDistributionAdvisee rgn, CacheProfile profile) {
      Scope otherScope = rgn.getAttributes().getScope();
      String result = null;

      // Verify both VMs are gateway-enabled or neither are. Note that since
      // this string is sent back to the caller, the 'other' and the 'my'
      // below are from the caller's point of view.
      final DistributedMember myId = rgn.getDistributionManager().getId();
      boolean otherCCEnabled = rgn.getAttributes().getConcurrencyChecksEnabled();
      boolean skipCheckForAccessor = skipCheckForAccessor(rgn, profile);
      boolean skipConcurrencyChecks = skipChecksForInternalRegion(rgn);
      boolean initializing = skipDuringInitialization(rgn);

      if (initializing) {
        this.skippedCompatibilityChecks = true;
      }

      if (!initializing && !skipCheckForAccessor && (rgn.getAttributes().getDataPolicy()
          .withPersistence() != profile.dataPolicy.withPersistence())) {
        // 45186: Do not allow a persistent replicate to be started if a
        // non-persistent replicate is running
        if (!rgn.getAttributes().getDataPolicy().withPersistence()) {
          result =
              String.format(
                  "Cannot create region %s DataPolicy withPersistence=true because another cache (%s) has the same region DataPolicy withPersistence=false. Persistent members must be started before non-persistent members",
                  regionPath, myId);
          skipConcurrencyChecks = true;
        } else {
          // make the new member turn on concurrency checks
          skipConcurrencyChecks = true;
        }
      }

      if (!initializing && !skipCheckForAccessor && !skipConcurrencyChecks
          && this.concurrencyChecksEnabled != otherCCEnabled) {
        result =
            String.format(
                "Cannot create region %s concurrency-checks-enabled=%s because another cache (%s) has the same region concurrency-checks-enabled=%s",
                regionPath, this.concurrencyChecksEnabled, myId, otherCCEnabled);
      }

      Set<String> otherGatewaySenderIds = ((LocalRegion) rgn).getGatewaySenderIds();
      Set<String> myGatewaySenderIds = profile.gatewaySenderIds;
      if (!otherGatewaySenderIds.equals(myGatewaySenderIds)) {
        if (!rgn.getFullPath().contains(DynamicRegionFactory.DYNAMIC_REGION_LIST_NAME)) {
          result =
              String.format(
                  "Cannot create Region %s with %s gateway sender ids because another cache has the same region defined with %s gateway sender ids",
                  this.regionPath, myGatewaySenderIds, otherGatewaySenderIds);
        }
      }

      Set<String> otherAsynEventQueueIds = ((LocalRegion) rgn).getVisibleAsyncEventQueueIds();
      Set<String> myAsyncEventQueueIds = profile.asyncEventQueueIds;
      if (!isLocalOrRemoteAccessor(rgn, profile)
          && !otherAsynEventQueueIds.equals(myAsyncEventQueueIds)) {
        result =
            String.format(
                "Cannot create Region %s with %s async event ids because another cache has the same region defined with %s async event ids",
                this.regionPath, myAsyncEventQueueIds, otherAsynEventQueueIds);
      }

      final PartitionAttributes pa = rgn.getAttributes().getPartitionAttributes();
      if (pa == null && profile.isPartitioned) {
        result =
            String.format(
                "Cannot create PartitionedRegion %s because another cache (%s) has the same region defined as a non PartitionedRegion.",
                this.regionPath, myId);
      } else if (pa != null && !profile.isPartitioned) {
        result =
            String.format(
                "Cannot create the non PartitionedRegion %s because another cache (%s) has a Partitioned Region defined with the same name.",
                this.regionPath, myId);
      } else if (profile.scope.isDistributed() && otherScope.isDistributed()) {
        // This check is somewhat unnecessary as all Partitioned Regions should have the same scope
        // due to the fact that Partitioned Regions do no support scope.
        if (profile.scope != otherScope) {
          result =
              String.format(
                  "Cannot create region %s with %s scope because another cache (%s) has same region with %s scope.",
                  this.regionPath, profile.scope, myId, otherScope);
        }
      }

      final boolean otherIsOffHeap = rgn.getAttributes().getOffHeap();

      boolean thisIsRemoteAccessor = false;
      if (!rgn.getAttributes().getDataPolicy().withStorage()
          || (pa != null && pa.getLocalMaxMemory() == 0)) {
        thisIsRemoteAccessor = true;
      }

      if (!isRemoteAccessor(profile) && !thisIsRemoteAccessor
          && profile.isOffHeap != otherIsOffHeap) {
        result =
            String.format(
                "Cannot create region %s with off-heap=%s because another cache (%s) has the same region with off-heap=%s.",
                this.regionPath, profile.isOffHeap, myId, otherIsOffHeap);
      }

      String cspResult = null;
      Map<String, CacheServiceProfile> myProfiles = ((LocalRegion) rgn).getCacheServiceProfiles();
      // Iterate and compare the remote CacheServiceProfiles to the local ones
      for (CacheServiceProfile remoteProfile : profile.cacheServiceProfiles) {
        CacheServiceProfile localProfile = myProfiles.get(remoteProfile.getId());
        if (localProfile == null) {
          cspResult = getMissingProfileMessage(remoteProfile, true);
        } else {
          cspResult = remoteProfile.checkCompatibility(rgn.getFullPath(), localProfile);
        }
        if (cspResult != null) {
          break;
        }
      }

      // If the comparison result is null, compare the local profiles to the remote ones. If there
      // are more local profiles than remote ones (meaning there are ones defined locally that are
      // not defined remotely), then compare those. This should produce an informative error message
      // (as opposed to returning something like 'the profiles don't match').
      if (cspResult == null) {
        if (myProfiles.size() > profile.cacheServiceProfiles.size()) {
          for (CacheServiceProfile localProfile : myProfiles.values()) {
            if (!profile.cacheServiceProfiles.stream()
                .anyMatch(remoteProfile -> remoteProfile.getId().equals(localProfile.getId()))) {
              cspResult = getMissingProfileMessage(localProfile, false);
              break;
            }
          }
        }
      }

      // If the comparison result is not null, set the final result.
      // Note: Be careful not to overwrite the final result with null in case it has already been
      // set in a previous compatibility check.
      if (cspResult != null) {
        result = cspResult;
      }

      if (logger.isDebugEnabled()) {
        logger.debug("CreateRegionProcessor.checkCompatibility: this={}; other={}; result={}", rgn,
            profile, result);
      }

      return result;
    }

    protected String getMissingProfileMessage(CacheServiceProfile profile,
        boolean existsInThisMember) {
      return profile.getMissingProfileMessage(existsInThisMember);
    }

    /**
     * When many members are started concurrently, it is possible that an accessor or non-version
     * generating replicate receives CreateRegionMessage before it is initialized, thus preventing
     * persistent members from starting. We skip compatibilityChecks if the region is not
     * initialized, and let other members check compatibility. If all members skipCompatabilit
     * checks, then the CreateRegionMessage should be retried. fixes #45186
     */
    private boolean skipDuringInitialization(CacheDistributionAdvisee rgn) {
      boolean skip = false;
      if (rgn instanceof LocalRegion) {
        LocalRegion lr = (LocalRegion) rgn;
        if (!lr.isInitialized()) {
          Set recipients = new CreateRegionProcessor(rgn).getRecipients();
          recipients.remove(getSender());
          if (!recipients.isEmpty()) {
            skip = true;
          }
        }
      }
      return skip;
    }

    /**
     * For internal regions skip concurrency-checks-enabled checks, since we will set it to true
     * after profile exchange if required.
     */
    private boolean skipChecksForInternalRegion(CacheDistributionAdvisee rgn) {
      boolean skip = false;
      if (rgn instanceof LocalRegion) {
        LocalRegion lr = (LocalRegion) rgn;
        skip = lr.isInternalRegion();
      }
      return skip;
    }

    /**
     * check for isLocalOrRemoteAccessor(CacheDistributionAdvisee, CacheProfile) and check if
     * DistributedRegion does not generate entry versions.
     */
    private boolean skipCheckForAccessor(CacheDistributionAdvisee rgn, CacheProfile profile) {
      boolean skip = false;
      if (rgn instanceof DistributedRegion) {
        DistributedRegion dr = (DistributedRegion) rgn;
        skip = !dr.getGenerateVersionTag();
      }
      return skip || isLocalOrRemoteAccessor(rgn, profile);
    }

    /**
     * @return true if profile being exchanged or region is an accessor i.e has no storage
     */
    protected static boolean isLocalOrRemoteAccessor(CacheDistributionAdvisee region,
        CacheProfile profile) {
      return isLocalAccessor(region) || isRemoteAccessor(profile);
    }

    protected static boolean isLocalAccessor(CacheDistributionAdvisee region) {
      if (!region.getAttributes().getDataPolicy().withStorage()) {
        return true;
      }
      if (region.getAttributes().getPartitionAttributes() != null
          && region.getAttributes().getPartitionAttributes().getLocalMaxMemory() == 0) {
        return true;
      }
      return false;
    }

    protected static boolean isRemoteAccessor(CacheProfile profile) {
      if (!profile.dataPolicy.withStorage()) {
        return true;
      }
      if (profile.isPartitioned) {
        PartitionProfile prProfile = (PartitionProfile) profile;
        if (prProfile.localMaxMemory == 0) {
          return true;
        }
      }
      return false;
    }

    @Override
    public void reset() {
      super.reset();
      this.regionPath = null;
      this.profile = null;
      this.processorId = -1;
    }

    @Override
    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
      super.fromData(in);
      this.regionPath = DataSerializer.readString(in);
      this.profile = (CacheProfile) DataSerializer.readObject(in);
      this.processorId = in.readInt();
      this.concurrencyChecksEnabled = in.readBoolean();
    }

    @Override
    public int getDSFID() {
      return CREATE_REGION_MESSAGE;
    }

    @Override
    public void toData(DataOutput out) throws IOException {
      super.toData(out);
      DataSerializer.writeString(this.regionPath, out);
      DataSerializer.writeObject(this.profile, out);
      out.writeInt(this.processorId);
      out.writeBoolean(this.concurrencyChecksEnabled);
    }

    @Override
    public String toString() {
      StringBuilder sb = new StringBuilder();
      sb.append("CreateRegionMessage (region='");
      sb.append(this.regionPath);
      sb.append("'; processorId=");
      sb.append(this.processorId);
      sb.append("; concurrencyChecksEnabled=").append(this.concurrencyChecksEnabled);
      sb.append("; profile=");
      sb.append(this.profile);
      sb.append(")");
      return sb.toString();
    }
  }

  public static class CreateRegionReplyMessage extends ReplyMessage {
    protected CacheProfile profile;
    protected ArrayList bucketProfiles;
    protected Object eventState;
    /**
     * Added to fix 42051. If the region is in the middle of being destroyed, return the destroyed
     * profile
     */
    protected PersistentMemberID destroyedId;
    protected boolean skippedCompatibilityChecks;

    long seqKeyForWan = -1;

    @Override
    public int getDSFID() {
      return CREATE_REGION_REPLY_MESSAGE;
    }

    @Override
    public boolean sendViaUDP() {
      return true;
    }

    @Override
    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
      super.fromData(in);
      if (in.readBoolean()) {
        this.profile = (CacheProfile) DataSerializer.readObject(in);
      }
      int size = in.readInt();
      if (size == 0) {
        this.bucketProfiles = null;
      } else {
        this.bucketProfiles = new ArrayList(size);
        for (int i = 0; i < size; i++) {
          RegionAdvisor.BucketProfileAndId bp = new RegionAdvisor.BucketProfileAndId();
          InternalDataSerializer.invokeFromData(bp, in);
          this.bucketProfiles.add(bp);
        }
      }
      if (in.readBoolean()) {
        this.eventState = EventStateHelper.deDataSerialize(in, false);
      }
      if (in.readBoolean()) {
        this.destroyedId = new PersistentMemberID();
        InternalDataSerializer.invokeFromData(this.destroyedId, in);
      }
      this.skippedCompatibilityChecks = in.readBoolean();
      this.seqKeyForWan = in.readLong();
    }

    @Override
    public void toData(DataOutput out) throws IOException {
      super.toData(out);
      out.writeBoolean(this.profile != null);
      if (this.profile != null) {
        DataSerializer.writeObject(this.profile, out);
      }
      if (this.bucketProfiles == null) {
        out.writeInt(0);
      } else {
        int size = this.bucketProfiles.size();
        out.writeInt(size);
        for (int i = 0; i < size; i++) {
          RegionAdvisor.BucketProfileAndId bp =
              (RegionAdvisor.BucketProfileAndId) this.bucketProfiles.get(i);
          InternalDataSerializer.invokeToData(bp, out);
        }
      }
      if (this.eventState != null) {
        out.writeBoolean(true);
        // The isHARegion flag is false here because
        // we currently only include the event state in the profile
        // for bucket regions.
        EventStateHelper.dataSerialize(out, (Map) eventState, false, getSender());
      } else {
        out.writeBoolean(false);
      }
      if (this.destroyedId != null) {
        out.writeBoolean(true);
        InternalDataSerializer.invokeToData(destroyedId, out);
      } else {
        out.writeBoolean(false);
      }
      out.writeBoolean(this.skippedCompatibilityChecks);
      out.writeLong(seqKeyForWan);
    }

    @Override
    public String toString() {
      StringBuilder buff = new StringBuilder();
      buff.append("CreateRegionReplyMessage");
      buff.append("(sender=").append(getSender());
      buff.append("; processorId=");
      buff.append(super.processorId);
      buff.append("; profile=");
      buff.append(this.profile);
      if (this.bucketProfiles != null) {
        buff.append("; bucketProfiles=");
        buff.append(this.bucketProfiles);
      }
      if (this.eventState != null) {
        buff.append("; eventState=<not null>");
      }
      buff.append("; skippedCompatibilityChecks=");
      buff.append(this.skippedCompatibilityChecks);
      buff.append("; seqKeyForWan=");
      buff.append(this.seqKeyForWan);
      if (this.getException() != null) {
        buff.append("; with exception {").append(getException().getMessage()).append("}");
      }
      buff.append(")");
      return buff.toString();
    }
  }
}