/*
 * 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.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.FixedPartitionAttributes;
import org.apache.geode.cache.FixedPartitionResolver;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.partition.PartitionNotAvailableException;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache.util.CacheWriterAdapter;
import org.apache.geode.distributed.DistributedMember;
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.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
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.logging.LogService;

public class PartitionedRegionHelper {
  private static final Logger logger = LogService.getLogger();

  /** 1 MB */
  static final long BYTES_PER_MB = 1024 * 1024;

  /**
   * The administrative region used for storing Partitioned Region meta data sub regions *
   */
  public static final String PR_ROOT_REGION_NAME = "__PR";

  /** Name of the DistributedLockService that PartitionedRegions used. */
  public static final String PARTITION_LOCK_SERVICE_NAME = "__PRLS";

  /** This is used to create bucket regions */
  static final String BUCKET_REGION_PREFIX = "_B_";

  /**
   * Time to wait for ownership (ms)
   * <p>
   * This should not be used normally. Internally, GemFire uses global locks to modify shared
   * meta-data and this property controls the delay before giving up trying to acquire a global lock
   */
  static final String VM_OWNERSHIP_WAIT_TIME_PROPERTY =
      DistributionConfig.GEMFIRE_PREFIX + "VM_OWNERSHIP_WAIT_TIME";

  /** Wait forever for ownership */
  static final long VM_OWNERSHIP_WAIT_TIME_DEFAULT = Long.MAX_VALUE;

  static final String MAX_PARTITIONED_REGION_ID = "MAX_PARTITIONED_REGION_ID";

  public static final int DEFAULT_WAIT_PER_RETRY_ITERATION = 100; // milliseconds

  public static final int DEFAULT_TOTAL_WAIT_RETRY_ITERATION = 60 * 60 * 1000; // milliseconds

  @Immutable
  public static final DataPolicy DEFAULT_DATA_POLICY = DataPolicy.PARTITION;

  @Immutable
  public static final Set ALLOWED_DATA_POLICIES;

  static final Object dlockMonitor = new Object();

  static {
    Set policies = new HashSet();
    policies.add(DEFAULT_DATA_POLICY);
    policies.add(DataPolicy.PERSISTENT_PARTITION);
    // policies.add(DataPolicy.NORMAL);
    ALLOWED_DATA_POLICIES = Collections.unmodifiableSet(policies);
  }

  /**
   * This function is used for cleaning the config meta data for the failed or closed
   * PartitionedRegion node.
   *
   * @param failedNode The failed PartitionedRegion Node
   * @param regionIdentifier The PartitionedRegion for which the cleanup is required
   * @param cache GemFire cache.
   */
  static void removeGlobalMetadataForFailedNode(Node failedNode, String regionIdentifier,
      InternalCache cache) {
    removeGlobalMetadataForFailedNode(failedNode, regionIdentifier, cache, true);
  }

  /**
   * This function is used for cleaning the config meta data for the failed or closed
   * PartitionedRegion node.
   *
   * @param failedNode The failed PartitionedRegion Node
   * @param regionIdentifier The PartitionedRegion for which the cleanup is required
   * @param cache GemFire cache.
   * @param lock True if this removal should acquire and release the RegionLock
   */
  static void removeGlobalMetadataForFailedNode(Node failedNode, String regionIdentifier,
      InternalCache cache, final boolean lock) {
    Region root = PartitionedRegionHelper.getPRRoot(cache, false);
    if (root == null) {
      return; // no partitioned region info to clean up
    }
    PartitionRegionConfig prConfig = (PartitionRegionConfig) root.get(regionIdentifier);
    if (null == prConfig || !prConfig.containsNode(failedNode)) {
      return;
    }

    final PartitionedRegion.RegionLock rl =
        PartitionedRegion.getRegionLock(regionIdentifier, cache);
    try {
      if (lock) {
        rl.lock();
      }
      prConfig = (PartitionRegionConfig) root.get(regionIdentifier);
      if (prConfig != null && prConfig.containsNode(failedNode)) {
        if (logger.isDebugEnabled()) {
          logger.debug("Cleaning up config for pr {} node {}", regionIdentifier, failedNode);
        }
        if ((prConfig.getNumberOfNodes() - 1) <= 0) {
          if (logger.isDebugEnabled()) {
            logger.debug("No nodes left but failed node {} destroying entry {} nodes {}",
                failedNode, regionIdentifier, prConfig.getNodes());
          }
          try {
            root.destroy(regionIdentifier);
          } catch (EntryNotFoundException e) {
            logger.warn(
                String.format("Got EntryNotFoundException in destroy Op for allPRRegion key, %s",
                    regionIdentifier),
                e);
          }
        } else {
          prConfig.removeNode(failedNode);
          if (prConfig.getNumberOfNodes() == 0) {
            root.destroy(regionIdentifier);
          } else {
            // We can't go backwards, or we potentially lose data
            root.put(regionIdentifier, prConfig);
          }
        }
      }
    } finally {
      if (lock) {
        rl.unlock();
      }
    }
  }

  /**
   * Return a region that is the root for all Partitioned Region metadata on this node
   */
  public static LocalRegion getPRRoot(final InternalCache cache) {
    return getPRRoot(cache, true);
  }

  /**
   * Return a region that is the root for all PartitionedRegion meta data on this Node. The main
   * administrative Regions contained within are <code>allPartitionedRegion</code> (Scope
   * DISTRIBUTED_ACK) and <code>bucket2Node</code> (Scope DISTRIBUTED_ACK) and dataStore regions.
   *
   * @return a GLOBLAL scoped root region used for PartitionedRegion administration
   */
  public static LocalRegion getPRRoot(final InternalCache cache, boolean createIfAbsent) {
    DistributedRegion root = (DistributedRegion) cache.getRegion(PR_ROOT_REGION_NAME, true);
    if (root == null) {
      if (!createIfAbsent) {
        return null;
      }
      if (logger.isDebugEnabled()) {
        logger.debug("Creating root Partitioned Admin Region {}",
            PartitionedRegionHelper.PR_ROOT_REGION_NAME);
      }
      AttributesFactory factory = new AttributesFactory();
      factory.setScope(Scope.DISTRIBUTED_ACK);
      factory.setDataPolicy(DataPolicy.REPLICATE);
      factory.addCacheListener(new FixedPartitionAttributesListener());
      if (Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "PRDebug")) {
        factory.addCacheListener(new CacheListenerAdapter() {
          @Override
          public void afterCreate(EntryEvent event) {
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "Create Event for allPR: key = {} oldVal = {} newVal = {} Op = {} origin = {} isNetSearch = {}",
                  event.getKey(), event.getOldValue(), event.getNewValue(), event.getOperation(),
                  event.getDistributedMember(), event.getOperation().isNetSearch());
            }
          }

          @Override
          public void afterUpdate(EntryEvent event) {
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "Update Event for allPR: key = {} oldVal = {} newVal = {} Op = {} origin = {} isNetSearch = {}",
                  event.getKey(), event.getOldValue(), event.getNewValue(), event.getOperation(),
                  event.getDistributedMember(), event.getOperation().isNetSearch());
            }
          }

          @Override
          public void afterDestroy(EntryEvent event) {
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "Destroy Event for allPR: key = {} oldVal = {} newVal = {} Op = {} origin = {} isNetSearch = {}",
                  event.getKey(), event.getOldValue(), event.getNewValue(), event.getOperation(),
                  event.getDistributedMember(), event.getOperation().isNetSearch());
            }
          }
        });

        factory.setCacheWriter(new CacheWriterAdapter() {
          @Override
          public void beforeUpdate(EntryEvent event) throws CacheWriterException {
            // the prConfig node list must advance (otherwise meta data becomes out of sync)
            final PartitionRegionConfig newConf = (PartitionRegionConfig) event.getNewValue();
            final PartitionRegionConfig oldConf = (PartitionRegionConfig) event.getOldValue();
            if (newConf != oldConf && !newConf.isGreaterNodeListVersion(oldConf)) {
              throw new CacheWriterException(
                  String.format(
                      "New PartitionedRegionConfig %s does not have newer version than previous %s",
                      new Object[] {newConf, oldConf}));
            }
          }
        });
      }

      RegionAttributes ra = factory.create();
      // Create anonymous stats holder for Partitioned Region meta data
      final HasCachePerfStats prMetaStatsHolder = new HasCachePerfStats() {
        @Override
        public CachePerfStats getCachePerfStats() {
          return new CachePerfStats(cache.getDistributedSystem(), "RegionStats-partitionMetaData");
        }
      };

      try {
        root = (DistributedRegion) cache.createVMRegion(PR_ROOT_REGION_NAME, ra,
            new InternalRegionArguments().setIsUsedForPartitionedRegionAdmin(true)
                .setInternalRegion(true).setCachePerfStatsHolder(prMetaStatsHolder));
        root.getDistributionAdvisor().addMembershipListener(new MemberFailureListener(cache));
      } catch (RegionExistsException ignore) {
        // we avoid this before hand, but yet we have to catch it
        root = (DistributedRegion) cache.getRegion(PR_ROOT_REGION_NAME, true);
      } catch (IOException ieo) {
        Assert.assertTrue(false, "IOException creating Partitioned Region root: " + ieo);
      } catch (ClassNotFoundException cne) {
        Assert.assertTrue(false, "ClassNotFoundExcpetion creating Partitioned Region root: " + cne);
      }
    }
    Assert.assertTrue(root != null,
        "Can not obtain internal Partitioned Region configuration root");
    return root;
  }

  // TODO rebalancing - this code was added here in the merge of -r22804:23093 from trunk
  // because of changes made on trunk that require this method, which was removed on
  // prRebalancing. It probably needs refactoring.
  // The idea here is to remove meta data from the partitioned region for a node that
  // has left the cache.
  // A couple options that didn't work
  // - remove metadata in region advisor for PR instead - this doesn't work because
  // the a member can close it's cache and then recreate the same region. Another member
  // might end up removing meta data after the region is recreated, leading to inconsistent metadata
  // - remove metadata on cache closure in the member that is closing - This didn't work because
  // we can't do region operations after isClosing is set to true (to remove metadata). Removing
  // metadata
  // before is closing is set to true results operations being silently ignored because of
  // inconsistent metadata
  // and regions.
  /**
   * Clean the config meta data for a DistributedMember which has left the DistributedSystem, one
   * PartitionedRegion at a time.
   */
  public static void cleanUpMetaDataOnNodeFailure(InternalCache cache,
      DistributedMember failedMemId) {
    try {
      if (cache == null || cache.getCancelCriterion().isCancelInProgress()) {
        return;
      }

      if (logger.isDebugEnabled()) {
        logger.debug("Cleaning PartitionedRegion meta data for memberId={}", failedMemId);
      }
      Region rootReg = PartitionedRegionHelper.getPRRoot(cache, false);
      if (rootReg == null) {
        return;
      }

      final ArrayList<String> ks = new ArrayList<String>(rootReg.keySet());
      if (ks.size() > 1) {
        Collections.shuffle(ks, PartitionedRegion.RANDOM);
      }
      for (String prName : ks) {
        try {
          cleanUpMetaDataForRegion(cache, prName, failedMemId, null);

        } catch (CancelException ignore) {
          // okay to ignore this - metadata will be cleaned up by cache close operation
        } catch (Exception e) {
          if (logger.isDebugEnabled()) {
            logger.debug("Got exception in cleaning up metadata. {}", e.getMessage(), e);
          }
        }
      }
    } catch (CancelException ignore) {
      // ignore
    }
  }

  public static void cleanUpMetaDataForRegion(final InternalCache cache, final String prName,
      final DistributedMember failedMemId, final Runnable postCleanupTask) {
    boolean runPostCleanUp = true;
    try {
      final PartitionRegionConfig prConf;
      Region rootReg = PartitionedRegionHelper.getPRRoot(cache, false);
      if (rootReg == null) {
        return;
      }
      try {
        prConf = (PartitionRegionConfig) rootReg.get(prName);
      } catch (EntryDestroyedException ignore) {
        return;
      }
      if (prConf == null) {
        // darrel says: I'm seeing an NPE in this code after pr->rem
        // merge
        // so I added this check and continue
        return;
      }
      Set<Node> nodeList = prConf.getNodes();
      if (nodeList == null) {
        return;
      }

      for (final Node node1 : nodeList) {
        if (cache.getCancelCriterion().isCancelInProgress()) {
          return;
        }
        if (node1.getMemberId().equals(failedMemId)) {
          // Do the cleanup in another thread so we don't have the advisor locked.
          // Fix for #45365, we don't schedule an asynchronous task until
          // we have determined the node to remove (Which includes the
          // serial number).
          cache.getDistributionManager().getPrMetaDataCleanupThreadPool().execute(new Runnable() {
            @Override
            public void run() {
              cleanPartitionedRegionMetaDataForNode(cache, node1, prConf, prName);
              if (postCleanupTask != null) {
                postCleanupTask.run();
              }
            }
          });
          runPostCleanUp = false;
          return;
        }
      }
    } finally {
      if (runPostCleanUp && postCleanupTask != null) {
        postCleanupTask.run();
      }
    }
  }

  /**
   * This is a function for cleaning the config meta data (both the configuration data and the
   * buckets) for a Node that hosted a PartitionedRegion
   */
  private static void cleanPartitionedRegionMetaDataForNode(InternalCache cache, Node node,
      PartitionRegionConfig prConf, String regionIdentifier) {
    if (logger.isDebugEnabled()) {
      logger.debug(
          "Cleaning PartitionedRegion meta data for node={} for Partitioned Region={} configuration={}",
          node, regionIdentifier, prConf);
    }
    PartitionedRegionHelper.removeGlobalMetadataForFailedNode(node, regionIdentifier, cache);

    if (logger.isDebugEnabled()) {
      logger.debug("Done Cleaning PartitionedRegion meta data for memberId={} for {}", node,
          regionIdentifier);
    }
  }

  /**
   * Runs hashCode() on given key producing a long value and then finds absolute value of the
   * modulus with bucketSize. For better key distribution, possibly use MD5 or SHA or any unique ID
   * generator for the hash function.
   *
   * @param pr the partitioned region on which to operate
   * @param key the key on which to determine the hash key
   * @return the bucket id the key hashes to
   */
  // private static int NOSIGN = 0x7fffffff;
  /*
   * public static int getHashKey(PartitionedObject key) { PartitionedRegion pRegion =
   * (PartitionedRegion)entryOp.getRegion(); RoutingResolver resolver =
   * pRegion.getRoutingResolver();
   *
   * int totalNumberOfBuckets = pRegion.getTotalNumberOfBuckets(); Object resolveKey = null; if
   * (resolver == null) { resolveKey = key; } else {
   *
   * //resolveKey = resolver.getPartitionKey(key); resolveKey =
   * resolver.getRoutingObject((EntryOperation)key); } int hc = resolveKey.hashCode(); int bucketId
   * = hc % totalNumberOfBuckets; // Force positive bucket ids only return Math.abs(bucketId); // We
   * should use the same hash code spreader as most other java.util hash tables. // h += ~(h << 9);
   * // h ^= (h >>> 14); // h += (h << 4); // h ^= (h >>> 10); // h &= NOSIGN; // return h %
   * totalNumberOfBuckets; }
   **/

  private static PartitionResolver getResolver(PartitionedRegion pr, Object key,
      Object callbackArgument) {
    // First choice is one associated with the region
    PartitionResolver result = pr.getPartitionResolver();
    if (result != null) {
      return result;
    }

    // Second is the key
    if (key != null && key instanceof PartitionResolver) {
      return (PartitionResolver) key;
    }

    // Third is the callback argument
    if (callbackArgument != null && callbackArgument instanceof PartitionResolver) {
      return (PartitionResolver) callbackArgument;
    }

    // There is no resolver.
    return null;
  }

  /**
   * Runs hashCode() on given key/routing object producing a long value and then finds absolute
   * value of the modulus with bucketSize. For better key distribution, possibly use MD5 or SHA or
   * any unique ID generator for the hash function.
   *
   * @param pr the partitioned region on which to operate
   * @param operation operation
   * @param key the key on which to determine the hash key
   * @param callbackArgument the callbackArgument is passed to <code>PartitionResolver</code> to get
   *        Routing object
   * @return the bucket id the key/routing object hashes to
   */
  public static int getHashKey(PartitionedRegion pr, Operation operation, Object key, Object value,
      Object callbackArgument) {
    // avoid creating EntryOperation if there is no resolver
    try {
      return getHashKey(null, pr, operation, key, value, callbackArgument);
    } catch (IllegalStateException e) { // bug #43651 - check for shutdown before throwing this
      pr.getCache().getCancelCriterion().checkCancelInProgress(e);
      pr.checkClosedOrDestroyed(); // GEODE-2282 - check for region destroyed before throwing it
      throw e;
    }
  }

  /**
   * Runs hashCode() on given key/routing object producing a long value and then finds absolute
   * value of the modulus with bucketSize. For better key distribution, possibly use MD5 or SHA or
   * any unique ID generator for the hash function.
   *
   * @param event entry event created for this entry operation
   * @return the bucket id the key/routing object hashes to
   */
  public static int getHashKey(EntryOperation event) {
    return getHashKey(event, null, null, null, null, null);
  }

  /**
   * Runs hashCode() on given key/routing object producing a long value and then finds absolute
   * value of the modulus with bucketSize. For better key distribution, possibly use MD5 or SHA or
   * any unique ID generator for the hash function.
   *
   * @param event entry event created for this entry operation; can be null
   * @param pr the partitioned region on which to operate
   * @param operation operation
   * @param key the key on which to determine the hash key
   * @param callbackArgument the callbackArgument is passed to <code>PartitionResolver</code> to get
   *        Routing object
   * @return the bucket id the key/routing object hashes to
   */
  private static int getHashKey(EntryOperation event, PartitionedRegion pr, Operation operation,
      Object key, Object value, Object callbackArgument) {
    // avoid creating EntryOperation if there is no resolver
    if (event != null) {
      pr = (PartitionedRegion) event.getRegion();
      key = event.getKey();
      callbackArgument = event.getCallbackArgument();
    }

    PartitionResolver resolver = getResolver(pr, key, callbackArgument);
    Object resolveKey = null;
    if (pr.isFixedPartitionedRegion()) {
      String partition = null;
      if (resolver instanceof FixedPartitionResolver) {
        Map<String, Integer[]> partitionMap = pr.getPartitionsMap();
        if (event == null) {
          event = new EntryOperationImpl(pr, operation, key, value, callbackArgument);
        }
        partition =
            ((FixedPartitionResolver) resolver).getPartitionName(event, partitionMap.keySet());
        if (partition == null) {
          Object[] prms = new Object[] {pr.getName(), resolver};
          throw new IllegalStateException(
              String.format("For region %s, partition resolver %s returned partition name null",
                  prms));
        }
        Integer[] bucketArray = partitionMap.get(partition);
        if (bucketArray == null) {
          Object[] prms = new Object[] {pr.getName(), partition};
          throw new PartitionNotAvailableException(
              String.format(
                  "For FixedPartitionedRegion %s, partition %s is not available on any datastore.",
                  prms));
        }
        int numBukets = bucketArray[1];
        resolveKey = (numBukets == 1) ? partition : resolver.getRoutingObject(event);
      } else if (resolver == null) {
        throw new IllegalStateException(
            String.format(
                "For FixedPartitionedRegion %s, FixedPartitionResolver is not available (neither through the partition attribute partition-resolver nor key/callbackArg implementing FixedPartitionResolver)",
                pr.getName()));
      } else {
        Object[] prms = new Object[] {pr.getName(), resolver};
        throw new IllegalStateException(
            String.format(
                "For FixedPartitionedRegion %s, Resolver defined %s is not an instance of FixedPartitionResolver",
                prms));
      }
      return assignFixedBucketId(pr, partition, resolveKey);
    } else {
      // Calculate resolveKey.
      if (resolver == null) {
        // no custom partitioning at all
        resolveKey = key;
        if (resolveKey == null) {
          throw new IllegalStateException("attempting to hash null");
        }
      } else {
        if (event == null) {
          event = new EntryOperationImpl(pr, operation, key, value, callbackArgument);
        }
        resolveKey = resolver.getRoutingObject(event);
        if (resolveKey == null) {
          throw new IllegalStateException(
              "The RoutingObject returned by PartitionResolver is null.");
        }
      }
      // Finally, calculate the hash.
      return getHashKey(pr, resolveKey);
    }
  }

  private static int assignFixedBucketId(PartitionedRegion pr, String partition,
      Object resolveKey) {
    int startingBucketID = 0;
    int partitionNumBuckets = 0;
    boolean isPartitionAvailable = pr.getPartitionsMap().containsKey(partition);
    Integer[] partitionDeatils = pr.getPartitionsMap().get(partition);
    if (isPartitionAvailable) {
      startingBucketID = partitionDeatils[0];
      partitionNumBuckets = partitionDeatils[1];

      int hc = resolveKey.hashCode();
      int bucketId = Math.abs(hc % partitionNumBuckets);
      int partitionBucketID = bucketId + startingBucketID;
      assert partitionBucketID != KeyInfo.UNKNOWN_BUCKET;
      return partitionBucketID;
    }
    List<FixedPartitionAttributesImpl> localFPAs = pr.getFixedPartitionAttributesImpl();

    if (localFPAs != null) {
      for (FixedPartitionAttributesImpl fpa : localFPAs) {
        if (fpa.getPartitionName().equals(partition)) {
          isPartitionAvailable = true;
          partitionNumBuckets = fpa.getNumBuckets();
          startingBucketID = fpa.getStartingBucketID();
          break;
        }
      }
    }

    if (!isPartitionAvailable) {
      List<FixedPartitionAttributesImpl> remoteFPAs =
          pr.getRegionAdvisor().adviseAllFixedPartitionAttributes();
      for (FixedPartitionAttributesImpl fpa : remoteFPAs) {
        if (fpa.getPartitionName().equals(partition)) {
          isPartitionAvailable = true;
          partitionNumBuckets = fpa.getNumBuckets();
          startingBucketID = fpa.getStartingBucketID();
          break;
        }
      }
    }

    if (partitionNumBuckets == 0) {
      if (isPartitionAvailable) {
        Object[] prms = new Object[] {pr.getName(), partition};
        throw new IllegalStateException(
            String.format(
                "For region %s, For partition %s partition-num-buckets is set to 0. Buckets cann not be created on this partition.",
                prms));
      }
    }

    if (!isPartitionAvailable) {
      Object[] prms = new Object[] {pr.getName(), partition};
      throw new PartitionNotAvailableException(
          String.format("For region %s, partition name %s is not available on any datastore.",
              prms));
    }
    int hc = resolveKey.hashCode();
    int bucketId = Math.abs(hc % partitionNumBuckets);
    int partitionBucketID = bucketId + startingBucketID;
    assert partitionBucketID != KeyInfo.UNKNOWN_BUCKET;
    return partitionBucketID;
  }

  public static int getHashKey(PartitionedRegion pr, Object routingObject) {
    return getHashKey(routingObject, pr.getTotalNumberOfBuckets());
  }

  public static int getHashKey(Object routingObject, int totalNumBuckets) {
    int hc = routingObject.hashCode();
    int bucketId = hc % totalNumBuckets;
    // Force positive bucket ids only
    return Math.abs(bucketId);
  }

  public static PartitionedRegion getPartitionedRegion(String prName, Cache cache) {
    Region region = cache.getRegion(prName);
    if (region != null) {
      if (region instanceof PartitionedRegion)
        return (PartitionedRegion) region;
    }
    return null;
  }

  public static boolean isBucketRegion(String fullPath) {
    return getBucketName(fullPath) != null;
  }

  /**
   * Find a ProxyBucketRegion by parsing the region fullPath
   *
   * @param fullPath full region path to parse
   * @param postInit true if caller should wait for bucket initialization to complete
   * @return ProxyBucketRegion as Bucket or null if not found
   * @throws PRLocallyDestroyedException if the PartitionRegion is locally destroyed
   */
  public static Bucket getProxyBucketRegion(Cache cache, String fullPath, boolean postInit)
      throws PRLocallyDestroyedException {
    if (cache == null) {
      // No cache
      return null;
    }
    // fullPath = /__PR/_B_1_10
    String bucketName = getBucketName(fullPath);
    if (bucketName == null) {
      return null;
    }

    String prid = getPRPath(bucketName);
    // PartitionedRegion region =
    // PartitionedRegion.getPRFromId(Integer.parseInt(prid));

    Region region;
    final InitializationLevel oldLevel = LocalRegion.setThreadInitLevelRequirement(ANY_INIT);
    try {
      region = cache.getRegion(prid);
    } finally {
      LocalRegion.setThreadInitLevelRequirement(oldLevel);
    }
    if (region == null || !(region instanceof PartitionedRegion)) {
      return null;
    }

    PartitionedRegion pr = (PartitionedRegion) region;

    int bid = getBucketId(bucketName);
    RegionAdvisor ra = (RegionAdvisor) pr.getDistributionAdvisor();
    if (postInit) {
      return ra.getBucketPostInit(bid);
    } else if (!ra.areBucketsInitialized()) {
      // While the RegionAdvisor may be available, it's bucket meta-data may not be constructed yet
      return null;
    } else {
      return ra.getBucket(bid);
    }

  }

  private static final String BUCKET_FULL_PATH_PREFIX =
      PR_ROOT_REGION_NAME + Region.SEPARATOR + BUCKET_REGION_PREFIX;

  /**
   * Get the bucket string by parsing the region fullPath
   *
   * @param bucketFullPath full region path to parse
   * @return the bucket string or null if no bucket string is present
   */
  public static String getBucketName(String bucketFullPath) {
    if (bucketFullPath == null || bucketFullPath.length() == 0) {
      return null;
    }
    int idxStartRoot = bucketFullPath.indexOf(BUCKET_FULL_PATH_PREFIX);
    // parse bucketString
    if (idxStartRoot != -1) {
      int idxEndRoot = idxStartRoot + PR_ROOT_REGION_NAME.length() + Region.SEPARATOR.length();
      return bucketFullPath.substring(idxEndRoot);
    }

    if (logger.isDebugEnabled()) {
      logger.debug("getBucketString no match fullPath={}", bucketFullPath);
    }
    return null;
  }

  public static String getBucketFullPath(String prFullPath, int bucketId) {
    String name = getBucketName(prFullPath, bucketId);
    if (name != null)
      return Region.SEPARATOR + PR_ROOT_REGION_NAME + Region.SEPARATOR + name;

    return null;

  }

  public static String escapePRPath(String prFullPath) {
    String escaped = prFullPath.replace("_", "__");
    escaped = escaped.replace(Region.SEPARATOR_CHAR, '_');
    return escaped;
  }


  public static final String TWO_SEPARATORS = Region.SEPARATOR + Region.SEPARATOR;

  public static String unescapePRPath(String escapedPath) {
    String path = escapedPath.replace('_', Region.SEPARATOR_CHAR);
    path = path.replace(TWO_SEPARATORS, "_");
    return path;
  }

  public static String getBucketName(String prPath, int bucketId) {
    return PartitionedRegionHelper.BUCKET_REGION_PREFIX
        + PartitionedRegionHelper.escapePRPath(prPath) + PartitionedRegion.BUCKET_NAME_SEPARATOR
        + bucketId;
  }


  /**
   * Returns the PR name give the bucketName (see getBucketName).
   */
  public static String getPRPath(String bucketName) {
    // bucketName = _B_PRNAME_10
    int pridIdx = PartitionedRegionHelper.BUCKET_REGION_PREFIX.length();
    int bidSepIdx = bucketName.lastIndexOf(PartitionedRegion.BUCKET_NAME_SEPARATOR);
    Assert.assertTrue(bidSepIdx > -1, "getProxyBucketRegion failed on " + bucketName);
    return unescapePRPath(bucketName.substring(pridIdx, bidSepIdx));
  }

  /**
   * Returns the bucket id gvien the bucketName (see getBucketName).
   */
  public static int getBucketId(String bucketName) {
    // bucketName = _B_PRNAME_10
    int bidSepIdx = bucketName.lastIndexOf(PartitionedRegion.BUCKET_NAME_SEPARATOR);
    String bid = bucketName.substring(bidSepIdx + 1);
    return Integer.parseInt(bid);
  }

  /**
   * This method returns true if the last region in provided fullPath is a sub-region else it
   * returns false. If fullPath is "/REGION1" it would return false and if it is "/REGION1/REGION2",
   * it would return true, which means that Region2 is a sub-region.
   *
   * @param fullPath full path of the region
   * @return true if given full path has sub-regions else return false
   */

  public static boolean isSubRegion(String fullPath) {
    boolean isSubRegion = false;
    if (null != fullPath) {
      int idx = fullPath.indexOf(Region.SEPARATOR, Region.SEPARATOR.length());
      if (idx >= 0)
        isSubRegion = true;
    }
    return isSubRegion;
  }

  /**
   * Utility method to print warning when nodeList in b2n region is found empty. This will signify
   * potential data loss scenario.
   *
   * @param bucketId Id of Bucket whose nodeList in b2n is empty.
   * @param callingMethod methodName of the calling method.
   */
  public static void logForDataLoss(PartitionedRegion partitionedRegion, int bucketId,
      String callingMethod) {
    if (!Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "PRDebug")) {
      return;
    }
    Region root = PartitionedRegionHelper.getPRRoot(partitionedRegion.getCache());
    // Region allPartitionedRegions = PartitionedRegionHelper.getPRConfigRegion(
    // root, partitionedRegion.getCache());
    PartitionRegionConfig prConfig =
        (PartitionRegionConfig) root.get(partitionedRegion.getRegionIdentifier());
    if (prConfig == null)
      return;

    Set members = partitionedRegion.getDistributionManager().getDistributionManagerIds();
    logger.warn(
        "DATALOSS (  {}  ) :: Size of nodeList After verifyBucketNodes for bucket ID,  {}  is 0",
        callingMethod, bucketId);
    logger.warn("DATALOSS (  {}  ) :: NodeList from prConfig,  {}",
        callingMethod, printCollection(prConfig.getNodes()));
    logger.warn("DATALOSS (  {}  ) :: Current Membership List,  {}",
        callingMethod, printCollection(members));
  }

  /**
   * Utility method to print a collection.
   *
   */
  public static String printCollection(Collection c) {
    if (c != null) {
      StringBuilder sb = new StringBuilder("[");
      Iterator itr = c.iterator();
      while (itr.hasNext()) {
        sb.append(itr.next());
        if (itr.hasNext()) {
          sb.append(", ");
        }
      }
      sb.append("]");
      return sb.toString();
    } else {
      return "[null]";
    }
  }

  public static FixedPartitionAttributesImpl getFixedPartitionAttributesForBucket(
      PartitionedRegion pr, int bucketId) {
    List<FixedPartitionAttributesImpl> localFPAs = pr.getFixedPartitionAttributesImpl();

    if (localFPAs != null) {
      for (FixedPartitionAttributesImpl fpa : localFPAs) {
        if (fpa.hasBucket(bucketId)) {
          return fpa;
        }
      }
    }

    List<FixedPartitionAttributesImpl> remoteFPAs =
        pr.getRegionAdvisor().adviseAllFixedPartitionAttributes();
    for (FixedPartitionAttributesImpl fpa : remoteFPAs) {
      if (fpa.hasBucket(bucketId)) {
        return fpa;
      }
    }
    Object[] prms = new Object[] {pr.getName(), bucketId};
    throw new PartitionNotAvailableException(
        String.format(
            "For FixedPartitionedRegion %s, Fixed partition is not defined for bucket id %s on any datastore",
            prms));
  }

  private static Set<String> getAllAvailablePartitions(PartitionedRegion region) {
    Set<String> partitionSet = new HashSet<String>();
    List<FixedPartitionAttributesImpl> localFPAs = region.getFixedPartitionAttributesImpl();
    if (localFPAs != null) {
      for (FixedPartitionAttributesImpl fpa : localFPAs) {
        partitionSet.add(fpa.getPartitionName());
      }
    }

    List<FixedPartitionAttributesImpl> remoteFPAs =
        region.getRegionAdvisor().adviseAllFixedPartitionAttributes();
    for (FixedPartitionAttributes fpa : remoteFPAs) {
      partitionSet.add(fpa.getPartitionName());
    }
    return Collections.unmodifiableSet(partitionSet);
  }

  public static Set<FixedPartitionAttributes> getAllFixedPartitionAttributes(
      PartitionedRegion region) {
    Set<FixedPartitionAttributes> fpaSet = new HashSet<FixedPartitionAttributes>();
    List<FixedPartitionAttributesImpl> localFPAs = region.getFixedPartitionAttributesImpl();
    if (localFPAs != null) {
      fpaSet.addAll(localFPAs);
    }
    List<FixedPartitionAttributesImpl> remoteFPAs =
        region.getRegionAdvisor().adviseAllFixedPartitionAttributes();
    fpaSet.addAll(remoteFPAs);
    return fpaSet;
  }

  private static class MemberFailureListener implements MembershipListener {

    InternalCache cache = null;

    MemberFailureListener(InternalCache cache) {
      this.cache = cache;
    }

    @Override
    public void memberJoined(DistributionManager distributionManager,
        InternalDistributedMember id) {

    }

    @Override
    public void memberDeparted(DistributionManager distributionManager,
        final InternalDistributedMember id, boolean crashed) {
      PartitionedRegionHelper.cleanUpMetaDataOnNodeFailure(cache, id);
    }

    @Override
    public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id,
        InternalDistributedMember whoSuspected, String reason) {}

    @Override
    public void quorumLost(DistributionManager distributionManager,
        Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {}

  }

  static class FixedPartitionAttributesListener extends CacheListenerAdapter {
    private static final Logger logger = LogService.getLogger();

    @Override
    public void afterCreate(EntryEvent event) {
      PartitionRegionConfig prConfig = (PartitionRegionConfig) event.getNewValue();
      if (!prConfig.getElderFPAs().isEmpty()) {
        updatePartitionMap(prConfig);
      }
    }

    @Override
    public void afterUpdate(EntryEvent event) {
      PartitionRegionConfig prConfig = (PartitionRegionConfig) event.getNewValue();
      if (!prConfig.getElderFPAs().isEmpty()) {
        updatePartitionMap(prConfig);
      }
    }

    private void updatePartitionMap(PartitionRegionConfig prConfig) {
      int prId = prConfig.getPRId();
      PartitionedRegion pr = null;

      try {
        pr = PartitionedRegion.getPRFromId(prId);
        if (pr != null) {
          Map<String, Integer[]> partitionMap = pr.getPartitionsMap();
          for (FixedPartitionAttributesImpl fxPrAttr : prConfig.getElderFPAs()) {
            partitionMap.put(fxPrAttr.getPartitionName(),
                new Integer[] {fxPrAttr.getStartingBucketID(), fxPrAttr.getNumBuckets()});
          }
        }
      } catch (PRLocallyDestroyedException e) {
        logger.debug("PRLocallyDestroyedException : Region ={} is locally destroyed on this node",
            prConfig.getPRId(), e);
      }
    }
  }
}