@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(regionPath);
        if (destroyingRegion != null) {
          destroyedId = destroyingRegion.getPersistentID();
        }

        lclRgn = (LocalRegion) cache.getRegion(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, regionPath), 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 = eventState;
        replyMsg.destroyedId = destroyedId;
        replyMsg.setProcessorId(processorId);
        replyMsg.setSender(dm.getId()); // for EventStateHelper.dataSerialize
        replyMsg.setRecipient(getSender());
        replyMsg.skippedCompatibilityChecks = skippedCompatibilityChecks;

        if (lclRgn != null && lclRgn.isUsedForPartitionedRegionBucket()) {
          replyMsg.seqKeyForWan = ((BucketRegion) lclRgn).getEventSeqNum().get();
        }
        if (replyException != null && !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(getSender());
        }
      }
    }
IllegalStateException is an anticipated reply exception. Receiving multiple replies with this
exception is normal.
geode-core/src/main/java/org/apache/geode/internal/cache/CreateRegionProcessor.java