/*
 * 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.CacheServerImpl.CACHE_SERVER_BIND_ADDRESS_NOT_AVAILABLE_EXCEPTION_MESSAGE;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.util.AbstractSet;
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.Queue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.client.internal.locator.SerializationHelper;
import org.apache.geode.cache.partition.PartitionListener;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LockNotHeldException;
import org.apache.geode.distributed.LockServiceDestroyedException;
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.DLockService;
import org.apache.geode.distributed.internal.locks.DistributedMemberLock;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.BucketProfileUpdateMessage;
import org.apache.geode.internal.cache.partitioned.DeposePrimaryBucketMessage;
import org.apache.geode.internal.cache.partitioned.DeposePrimaryBucketMessage.DeposePrimaryBucketResponse;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.util.StopWatch;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Specialized {@link CacheDistributionAdvisor} for {@link BucketRegion BucketRegions}. The
 * <code>BucketAdvisor</code> is owned by a {@link ProxyBucketRegion} and may outlive a
 * <code>BucketRegion</code>.
 */
@SuppressWarnings("synthetic-access")
public class BucketAdvisor extends CacheDistributionAdvisor {
  private static final Logger logger = LogService.getLogger();

  private static final boolean ENFORCE_SAFE_CLOSE = false;
  // TODO: Boolean.getBoolean("gemfire.BucketAdvisor.debug.enforceSafeClose");

  /** Reference to the InternalDistributedMember that is primary. */
  private final AtomicReference<InternalDistributedMember> primaryMember = new AtomicReference<>();

  /**
   * Advice requests for {@link #adviseProfileUpdate()} delegate to the partitioned region's
   * <code>RegionAdvisor</code> to include members with {@link ProxyBucketRegion}s as well as real
   * {@link BucketRegion}s.
   */
  private final RegionAdvisor regionAdvisor;

  private final BucketRedundancyTracker redundancyTracker;

  /**
   * The bucket primary will be holding this distributed lock. Protected by synchronized(this).
   */
  private DistributedMemberLock primaryLock;

  // private static final byte MASK_HOSTING = 1; // 0001
  // private static final byte MASK_VOLUNTEERING = 2; // 0010
  // private static final byte MASK_OTHER_PRIMARY = 4; // 0100
  // private static final byte MASK_IS_PRIMARY = 8; // 1000

  private static final byte NO_PRIMARY_NOT_HOSTING = 0; // 0000_0000
  private static final byte NO_PRIMARY_HOSTING = 1; // 0000_0001
  private static final byte OTHER_PRIMARY_NOT_HOSTING = 4; // 0000_0100
  private static final byte OTHER_PRIMARY_HOSTING = 5; // 0000_0101
  private static final byte VOLUNTEERING_HOSTING = 3; // 0000_0011
  private static final byte BECOMING_HOSTING = 15; // 0000_1111
  private static final byte IS_PRIMARY_HOSTING = 9; // 0000_1001
  private static final byte CLOSED = 16; // 0001_0000

  /**
   * The current state of this BucketAdvisor which tracks which member is primary and whether or not
   * this member is hosting a real Bucket.
   */
  private volatile byte primaryState = NO_PRIMARY_NOT_HOSTING;

  /**
   * This delegate handles all volunteering for primary status. Lazily created. Protected by
   * synchronization(this).
   */
  private VolunteeringDelegate volunteeringDelegate;

  /**
   * A random number generator
   *
   * @see #getPreferredNode()
   */
  @Immutable
  private static final Random myRand = new Random();

  /**
   * A read/write lock to prevent making this bucket not primary while a write is in progress on the
   * bucket.
   */
  private final ReadWriteLock primaryMoveReadWriteLock;
  private final Lock primaryMoveReadLock;
  private final Lock primaryMoveWriteLock;

  /**
   * The advisor for the bucket region that we are colocated with, if this region is a colocated
   * region.
   */
  private final BucketAdvisor parentAdvisor;

  /**
   * The member that is responsible for choosing the primary for this bucket. While this field is
   * set and this member exists, this bucket won't try to become primary.
   */
  private volatile InternalDistributedMember primaryElector;

  private volatile BucketProfile localProfile;

  private volatile boolean everHadPrimary = false;

  private BucketAdvisor startingBucketAdvisor;

  private volatile boolean hasBecomePrimary = false;

  private final PartitionedRegion pRegion;

  final ConcurrentMap<String, Boolean> destroyedShadowBuckets = new ConcurrentHashMap<>();

  /**
   * Constructs a new BucketAdvisor for the Bucket owned by RegionAdvisor.
   *
   * @param bucket the bucket to provide metadata and advice for
   * @param regionAdvisor advisor for the PartitionedRegion
   */
  private BucketAdvisor(Bucket bucket, RegionAdvisor regionAdvisor) {
    super(bucket);
    this.regionAdvisor = regionAdvisor;
    pRegion = this.regionAdvisor.getPartitionedRegion();
    redundancyTracker =
        new BucketRedundancyTracker(pRegion.getRedundantCopies(), pRegion.getRedundancyTracker());
    parentAdvisor = getParentAdvisor(bucket.getId());

    if (parentAdvisor == null) {
      primaryMoveReadWriteLock = new ReentrantReadWriteLock();
    } else {
      primaryMoveReadWriteLock = parentAdvisor.primaryMoveReadWriteLock;
    }
    primaryMoveReadLock = primaryMoveReadWriteLock.readLock();
    primaryMoveWriteLock = primaryMoveReadWriteLock.writeLock();
  }

  public static BucketAdvisor createBucketAdvisor(Bucket bucket, RegionAdvisor regionAdvisor) {
    BucketAdvisor advisor = new BucketAdvisor(bucket, regionAdvisor);
    advisor.initialize();
    return advisor;
  }

  private BucketAdvisor getParentAdvisor(int bucketId) {
    PartitionedRegion colocatedRegion = ColocationHelper.getColocatedRegion(pRegion);
    if (colocatedRegion != null) {
      if (colocatedRegion.isFixedPartitionedRegion()) {
        List<FixedPartitionAttributesImpl> fpas = colocatedRegion.getFixedPartitionAttributesImpl();
        if (fpas != null) {
          for (FixedPartitionAttributesImpl fpa : fpas) {
            if (fpa.hasBucket(bucketId)) {
              return colocatedRegion.getRegionAdvisor().getBucketAdvisor(fpa.getStartingBucketID());
            }
          }
        }
      } else {
        return colocatedRegion.getRegionAdvisor().getBucketAdvisor(bucketId);
      }
    }
    return null;
  }

  private void assignStartingBucketAdvisorIfFixedPartitioned() {
    if (startingBucketAdvisor != null) {
      // already assigned
      return;
    }
    if (pRegion.isFixedPartitionedRegion()) {
      List<FixedPartitionAttributesImpl> fpas = pRegion.getFixedPartitionAttributesImpl();
      if (fpas != null) {
        int bucketId = getBucket().getId();
        for (FixedPartitionAttributesImpl fpa : fpas) {
          if (fpa.hasBucket(bucketId) && bucketId != fpa.getStartingBucketID()) {
            startingBucketAdvisor = regionAdvisor.getBucketAdvisor(fpa.getStartingBucketID());
            break;
          }
        }
      }
    }
  }

  /**
   * Returns the lock that prevents the primary from moving while active writes are in progress.
   * This should be locked before checking if the local bucket is primary.
   *
   * @return the lock for in-progress write operations
   */
  public Lock getPrimaryMoveReadLock() {
    return primaryMoveReadLock;
  }

  /**
   * Try to lock the primary bucket to make sure no operation is on-going at current bucket.
   *
   */
  void tryLockIfPrimary() {
    if (isPrimary()) {
      try {
        primaryMoveWriteLock.lock();
      } finally {
        primaryMoveWriteLock.unlock();
      }
    }
  }

  /**
   * Makes this <code>BucketAdvisor</code> give up being a primary and become a secondary. Does
   * nothing if not currently the primary.
   *
   * @return true if this advisor has been deposed as primary
   */
  public boolean deposePrimary() {
    if (isPrimary()) {
      primaryMoveWriteLock.lock();
      boolean needToSendProfileUpdate = false;
      try {
        removePrimary(getDistributionManager().getId());
        synchronized (this) {
          if (!isPrimary()) {
            // releasePrimaryLock();
            needToSendProfileUpdate = true;
            return true;
          } else {
            return false; // failed for some reason
          }
        }
      } finally {
        primaryMoveWriteLock.unlock();
        if (needToSendProfileUpdate) {
          sendProfileUpdate();
        }
      }
    } else {
      sendProfileUpdate();
      return true;
    }
  }

  /**
   * This calls deposePrimary on every colocated child that is directly colocated to this bucket's
   * PR. Those each in turn do the same to their child buckets and so on before returning. Each
   * depose will send a dlock release message to the grantor, wait for reply, and then also send a
   * profile update.
   * <p>
   * Caller must synchronize on this BucketAdvisor.
   *
   */
  private void deposePrimaryForColocatedChildren() {
    boolean deposedChildPrimaries = true;
    List<PartitionedRegion> colocatedChildPRs = ColocationHelper.getColocatedChildRegions(pRegion);
    if (colocatedChildPRs != null) {
      for (PartitionedRegion pr : colocatedChildPRs) {
        Bucket b = pr.getRegionAdvisor().getBucket(getBucket().getId());
        if (b != null) {
          BucketAdvisor ba = b.getBucketAdvisor();
          deposedChildPrimaries = ba.deposePrimary() && deposedChildPrimaries;
          if (b instanceof BucketRegionQueue) {
            BucketRegionQueue brq = (BucketRegionQueue) b;
            brq.decQueueSize(brq.size());
            brq.incSecondaryQueueSize(brq.size());
          }
        }
      }
    }
  }

  private void deposeOtherPrimaryBucketForFixedPartition() {
    boolean deposedOtherPrimaries = true;
    int bucketId = getBucket().getId();
    List<FixedPartitionAttributesImpl> fpas = pRegion.getFixedPartitionAttributesImpl();
    if (fpas != null) {
      for (FixedPartitionAttributesImpl fpa : fpas) {
        if (fpa.getStartingBucketID() == bucketId) {
          for (int i = (bucketId + 1); i <= fpa.getLastBucketID(); i++) {
            Bucket b = regionAdvisor.getBucket(i);
            if (b != null) {
              BucketAdvisor ba = b.getBucketAdvisor();
              deposedOtherPrimaries = ba.deposePrimary() && deposedOtherPrimaries;
            }
          }
        }
      }
    }
  }

  void removeBucket() {
    setHosting(false);
  }

  /**
   * Return (and possibly choose) a thread-sticky member from whose data store this bucket's values
   * should be read
   *
   * @return member to use for reads, null if none available
   */
  public InternalDistributedMember getPreferredNode() {

    if (isHosting()) {
      getPartitionedRegionStats().incPreferredReadLocal();
      return getDistributionManager().getId();
    }

    Profile[] locProfiles = profiles; // volatile read
    if (locProfiles.length == 0) {
      return null;
    }

    Profile selectedProfile = selectNotInitializingProfile(locProfiles);

    if (selectedProfile == null) {
      return null;
    }

    getPartitionedRegionStats().incPreferredReadRemote();
    return selectedProfile.peerMemberId;
  }

  /**
   * Get the random Profile that is not initializing BucketProfile.
   *
   */
  private Profile selectNotInitializingProfile(Profile[] inProfiles) {
    int index = 0;
    int offset = 0;
    if (inProfiles.length > 1) {
      // Pick random offset.
      offset = myRand.nextInt(inProfiles.length);
    }

    for (int i = 0; i < inProfiles.length; i++) {
      index = (offset + i) % inProfiles.length;
      BucketProfile bp = (BucketProfile) inProfiles[index];
      if (!bp.isInitializing) {
        return inProfiles[index];
      }
    }
    return null;
  }

  /**
   * Returns the thread-safe queue of primary volunteering tasks for the parent Partitioned Region.
   *
   * @return the queue of primary volunteering tasks
   */
  private Queue<Runnable> getVolunteeringQueue() {
    return regionAdvisor.getVolunteeringQueue();
  }

  /**
   * Returns the semaphore which controls the number of threads allowed to consume from the
   * {@link #getVolunteeringQueue volunteering queue}.
   *
   * @return the semaphore which controls the number of volunteering threads
   */
  private Semaphore getVolunteeringSemaphore() {
    return regionAdvisor.getVolunteeringSemaphore();
  }

  /**
   * Returns the PartitionedRegionStats.
   *
   * @return the PartitionedRegionStats
   */
  private PartitionedRegionStats getPartitionedRegionStats() {
    return regionAdvisor.getPartitionedRegionStats();
  }

  /**
   * Concurrency: protected by synchronizing on *this*
   */
  @Override
  protected void profileCreated(Profile profile) {
    regionAdvisor.incrementBucketCount(profile);
    super.profileCreated(profile);
    if (updateRedundancy() > 0) {
      // wake up any threads in waitForRedundancy or waitForPrimary
      notifyAll();
    }
    regionAdvisor.updateBucketStatus(getBucket().getId(), profile.peerMemberId, false);
    if (logger.isDebugEnabled()) {
      logger.debug("Profile added {} Profile : {}", getBucket().getFullPath(), profile);
    }
    synchronized (this) {
      updateServerBucketProfile();
    }
  }

  /**
   * Concurrency: protected by synchronizing on *this*
   */
  @Override
  protected void profileUpdated(Profile profile) {
    super.profileUpdated(profile);
    if (updateRedundancy() > 0) {
      // wake up any threads in waitForRedundancy or waitForPrimary
      notifyAll();
    }
    regionAdvisor.updateBucketStatus(getBucket().getId(), profile.peerMemberId, false);

    if (logger.isDebugEnabled()) {
      logger.debug("Profile updated {} Profile : {}", getBucket().getFullPath(), profile);
    }
    synchronized (this) {
      updateServerBucketProfile();
    }
  }

  /**
   * Concurrency: protected by synchronizing on *this*
   */
  @Override
  protected void profileRemoved(Profile profile) {
    if (profile != null) {
      regionAdvisor.updateBucketStatus(getBucket().getId(),
          profile.getDistributedMember(), true);
      regionAdvisor.decrementsBucketCount(profile);
    }
    updateRedundancy();

    if (logger.isDebugEnabled()) {
      logger.debug("Profile removed {} the member lost {} Profile : {}", getBucket().getFullPath(),
          profile != null ? profile.getDistributedMember() : null, profile);
    }
    synchronized (this) {
      updateServerBucketProfile();
    }
  }

  @Override
  public boolean shouldSyncForCrashedMember(InternalDistributedMember id) {
    BucketProfile profile = (BucketProfile) getProfile(id);
    return (profile != null) && (profile.isPrimary);
  }

  @Override
  public DistributedRegion getRegionForDeltaGII() {
    DistributedRegion result = super.getRegionForDeltaGII();
    if (result == null && getAdvisee() instanceof ProxyBucketRegion) {
      result = ((ProxyBucketRegion) getAdvisee()).getHostedBucketRegion();
    }
    return result;
  }


  BucketAdvisor getParentAdvisor() {
    return parentAdvisor;
  }

  boolean getHasBecomePrimary() {
    return hasBecomePrimary;
  }

  /**
   * Called by the RegionAdvisor.profileRemoved, this method tests to see if the missing member is
   * the primary elector for this bucket.
   *
   * We can't call this method from BucketAdvisor.profileRemoved, because the primaryElector may not
   * actually host the bucket.
   *
   */
  public void checkForLostPrimaryElector(Profile profile) {
    // If the member that went away was in the middle of creating
    // the bucket, finish the bucket creation.
    ProfileId elector = primaryElector;
    if (elector != null && elector.equals(profile.getDistributedMember())) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "Bucket {} lost the member responsible for electing the primary. Finishing bucket creation",
            getBucket().getFullPath());
      }
      primaryElector = getBucket().getDistributionManager().getId();
      getBucket().getDistributionManager().getExecutors().getWaitingThreadPool().execute(
          () -> getBucket().getPartitionedRegion().getRedundancyProvider()
              .finishIncompleteBucketCreation(getBucket().getId()));
    }
  }

  /**
   * Only allows profiles that actually hosting this bucket. If the profile is primary, then
   * primaryMember will be set to that member but only if we are not already the primary.
   *
   * @param profile the profile to add (must be a BucketProfile)
   * @param forceProfile true will force profile to be added even if member is not in distributed
   *        view
   *
   * @see #adviseProfileUpdate()
   */
  @Override
  public synchronized boolean putProfile(Profile profile, boolean forceProfile) {
    assert profile instanceof BucketProfile;
    BucketProfile bp = (BucketProfile) profile;

    // Only hosting buckets will be initializing, the isInitializing boolean is to
    // allow for early entry into the advisor for GII purposes
    if (!bp.isHosting && !bp.isInitializing) {
      if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
        logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE, "BucketAdvisor#putProfile early out");
      }
      return false; // Do not allow introduction of proxy profiles, they don't provide anything
                    // useful
      // isHosting = false, isInitializing = false
    }
    if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE)) {
      logger.trace(LogMarker.DISTRIBUTION_ADVISOR_VERBOSE,
          "BucketAdvisor#putProfile profile=<{}> force={}; profile = {}", profile, forceProfile,
          bp);
    }
    // isHosting = false, isInitializing = true
    // isHosting = true, isInitializing = false
    // isHosting = true, isInitializing = true... (false state)

    final boolean applied;
    synchronized (this) {
      // force new membership version in the advisor so that the
      // state flush mechanism can capture any updates to the bucket
      // MIN_VALUE is intended as a somewhat unique value for potential debug purposes
      profile.initialMembershipVersion = Long.MIN_VALUE;
      applied = super.putProfile(profile, forceProfile);
      // skip following block if isPrimary to avoid race where we process late
      // arriving OTHER_PRIMARY profile after we've already become primary
      if (applied && !isPrimary()) {
        if (bp.isPrimary) {
          setPrimaryMember(bp.getDistributedMember());
        } else {
          notPrimary(bp.getDistributedMember());
        }
      } // if: !isPrimary

    } // synchronized
    return applied;
  }

  private static <E> Set<E> newSetFromMap(Map<E, Boolean> map) {
    if (map.isEmpty()) {
      return new SetFromMap<>(map);
    }
    throw new IllegalArgumentException();
  }

  @SuppressWarnings("NullableProblems")
  private static class SetFromMap<E> extends AbstractSet<E> implements Serializable {
    private static final long serialVersionUID = 2454657854757543876L;

    // must named as it, to pass serialization compatibility test.
    private final Map<E, Boolean> m;

    private transient Set<E> backingSet;

    SetFromMap(final Map<E, Boolean> map) {
      super();
      m = map;
      backingSet = map.keySet();
    }

    @Override
    public boolean equals(Object object) {
      return backingSet.equals(object);
    }

    @Override
    public int hashCode() {
      return backingSet.hashCode();
    }

    @Override
    public boolean add(E object) {
      return m.put(object, Boolean.TRUE) == null;
    }

    @Override
    public void clear() {
      m.clear();
    }

    @Override
    public String toString() {
      return backingSet.toString();
    }

    @Override
    public boolean contains(Object object) {
      return backingSet.contains(object);
    }

    @Override
    public boolean containsAll(Collection<?> collection) {
      return backingSet.containsAll(collection);
    }

    @Override
    public boolean isEmpty() {
      return m.isEmpty();
    }

    @Override
    public boolean remove(Object object) {
      return m.remove(object) != null;
    }

    @Override
    public boolean retainAll(Collection<?> collection) {
      return backingSet.retainAll(collection);
    }

    @Override
    public Object[] toArray() {
      return backingSet.toArray();
    }

    @Override
    public <T> T[] toArray(T[] contents) {
      return backingSet.toArray(contents);
    }

    @Override
    public Iterator<E> iterator() {
      return backingSet.iterator();
    }

    @Override
    public int size() {
      return m.size();
    }

    private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
      stream.defaultReadObject();
      backingSet = m == null ? Collections.emptySet() : m.keySet();
    }
  }

  /**
   * repopulates the RegionAdvisor's location information for this bucket
   */
  private void updateServerBucketProfile() {
    int bucketId = getBucket().getId();
    Set<ServerBucketProfile> serverProfiles =
        newSetFromMap(new HashMap<>());
    for (Profile p : profiles) {
      if (p instanceof ServerBucketProfile) {
        serverProfiles.add((ServerBucketProfile) p);
      }
    }
    regionAdvisor.setClientBucketProfiles(bucketId, serverProfiles);
  }

  /**
   * Only for local profile.
   *
   */
  public synchronized void updateServerBucketProfile(BucketProfile p) {
    localProfile = p;
  }

  public BucketProfile getLocalProfile() {
    return localProfile;
  }

  @Override
  public boolean removeId(ProfileId memberId, boolean crashed, boolean destroyed,
      boolean fromMembershipListener) {
    boolean hadBucketRegion = super.removeId(memberId, crashed, destroyed, fromMembershipListener);
    if (hadBucketRegion) {
      // do NOT call notPrimary under synchronization
      try {
        notPrimary((InternalDistributedMember) memberId);
      } catch (CancelException e) {
        // must be closing the cache - no need to try to become primary
      }
    }
    return hadBucketRegion;
  }

  /**
   * Removes the profile for the specified member. If that profile is marked as primary, this will
   * call {@link #notPrimary(InternalDistributedMember)}.
   *
   * @param memberId the member to remove the profile for
   * @param serialNum specific serial number to remove
   * @return true if a matching profile for the member was found
   */
  @Override
  public boolean removeIdWithSerial(InternalDistributedMember memberId, int serialNum,
      boolean regionDestroyed) {
    boolean hadBucketRegion = super.removeIdWithSerial(memberId, serialNum, regionDestroyed);
    if (hadBucketRegion) {
      // do NOT call notPrimary under synchronization
      notPrimary(memberId);
    }
    forceNewMembershipVersion();
    return hadBucketRegion;
  }

  @Override
  public Set<InternalDistributedMember> adviseProfileExchange() {
    // delegate up to RegionAdvisor to include members that might have
    // ProxyBucketRegion without a real BucketRegion
    Assert.assertTrue(regionAdvisor.isInitialized());
    return regionAdvisor.adviseBucketProfileExchange();
  }

  @Override
  public Set<InternalDistributedMember> adviseProfileUpdate() {
    // delegate up to RegionAdvisor to include members that might have
    // ProxyBucketRegion without a real BucketRegion
    return regionAdvisor.adviseGeneric();
  }

  /**
   * Sets hosting to false and returns without closing. Calling closeAdvisor will actually close
   * this advisor.
   */
  @Override
  public void close() {
    // ok, so BucketRegion extends DistributedRegion which calls close on the
    // advisor, BUT the ProxyBucketRegion truly owns the advisor and only it
    // should be able to close the advisor...
    //
    // if a BucketRegion closes, it will call this method and we want to change
    // our state to NOT_HOSTING
    //
    // see this.closeAdvisor()
    setHosting(false);
  }

  /**
   * Blocks until there is a known primary and return that member, but only if there are real bucket
   * regions that exist. If there are no real bucket regions within the distribution config's
   * member-timeout setting * 3 (time required to eject a member) + 15000, then this returns null.
   *
   * kbanks: reworked this method to avoid JIT issue #40639
   *
   * @return the member who is primary for this bucket
   */
  public InternalDistributedMember getPrimary() {
    InternalDistributedMember primary = getExistingPrimary();
    if (primary == null) {
      primary = waitForNewPrimary();
    }
    return primary;
  }

  /**
   * This method was split out from getPrimary() due to bug #40639 and is only intended to be called
   * from within that method.
   *
   * @see #getPrimary()
   * @return the existing primary (if it is still in the view) otherwise null
   */
  private InternalDistributedMember getExistingPrimary() {
    return basicGetPrimaryMember();
  }

  /**
   * This method was split out from getPrimary() due to bug #40639 and is only intended to be called
   * from within that method.
   *
   * @see #getPrimary()
   * @return the new primary
   */
  private InternalDistributedMember waitForNewPrimary() {
    DistributionManager dm = regionAdvisor.getDistributionManager();
    DistributionConfig config = dm.getConfig();
    // failure detection period
    long timeout = config.getMemberTimeout() * 3L;
    // plus time for a new member to become primary
    timeout += Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "BucketAdvisor.getPrimaryTimeout",
        15000L);
    return waitForPrimaryMember(timeout);
  }

  /**
   * Marks member as not primary. Initiates volunteerForPrimary if this member is hosting a real
   * bucket. This method does nothing if the member parameter is the current member.
   *
   * @param member the member who is not primary
   */
  public void notPrimary(InternalDistributedMember member) {
    // Fix for 43569. Only the deposePrimary call should
    // make the local member drop the primary lock.
    if (!member.equals(getDistributionManager().getId())) {
      removePrimary(member);
    }
  }

  /**
   * Marks member as not primary. Initiates volunteerForPrimary if this member is hosting a real
   * bucket.
   *
   * @param member the member who is not primary
   */
  private void removePrimary(InternalDistributedMember member) {
    boolean needToVolunteerForPrimary = false;
    if (!isClosed()) { // hole: requestPrimaryState not hosting
      initializationGate();
    }
    boolean lostPrimary = false;
    try {
      synchronized (this) {
        boolean wasPrimary = isPrimary() && getDistributionManager().getId().equals(member);
        final InternalDistributedMember currentPrimary = primaryMember.get();
        if (currentPrimary != null && currentPrimary.equals(member)) {
          if (logger.isDebugEnabled()) {
            logger.debug("[BucketAdvisor.notPrimary] {} for {}", member, this);
          }
          primaryMember.set(null);
        } else {
          return;
        }

        if (isClosed()) {
          // possibly closed if caller comes from outside this advisor
          return; // return quietly
        }
        // member is primary... need to change state to NO_PRIMARY_xxx
        if (isHosting()) {
          requestPrimaryState(NO_PRIMARY_HOSTING);
          if (pRegion.isFixedPartitionedRegion()) {
            InternalDistributedMember primaryMember =
                regionAdvisor.adviseFixedPrimaryPartitionDataStore(getBucket().getId());
            needToVolunteerForPrimary = primaryMember == null || primaryMember.equals(member);
          } else {
            needToVolunteerForPrimary = true;
          }
        } else {
          requestPrimaryState(NO_PRIMARY_NOT_HOSTING);
        }
        if (wasPrimary) {
          lostPrimary = true;
        }

        // try to fix yet another cause of bug 36881
        findAndSetPrimaryMember();
      }
    } finally {
      if (lostPrimary) {
        invokeAfterSecondaryInPartitionListeners();
        Bucket br = regionAdvisor.getBucket(getBucket().getId());
        if (br instanceof BucketRegion) {
          ((BucketRegion) br).beforeReleasingPrimaryLockDuringDemotion();
        }

        releasePrimaryLock();
        // this was a deposePrimary call so we need to depose children as well
        deposePrimaryForColocatedChildren();

        if (pRegion.isFixedPartitionedRegion()) {
          deposeOtherPrimaryBucketForFixedPartition();
        }
      }
    }

    if (needToVolunteerForPrimary) {
      volunteerForPrimary();
    }
  }

  /**
   * Returns the ProxyBucketRegion which owns this advisor.
   *
   * @return the ProxyBucketRegion which owns this advisor
   */
  public ProxyBucketRegion getProxyBucketRegion() {
    return (ProxyBucketRegion) getAdvisee();
  }

  /**
   * Actually close this advisor for real. Called by ProxyBucketRegion only. Calling this method
   * actually closes this advisor whereas {@link #close()} only sets hosting to false.
   */
  void closeAdvisor() {
    boolean wasPrimary;
    synchronized (this) {
      if (isClosed()) {
        return;
      }
      wasPrimary = isPrimary();
      super.close();
      requestPrimaryState(CLOSED);
      redundancyTracker.closeBucket();
      localProfile = null;
    }
    if (wasPrimary) {
      releasePrimaryLock();
    }
  }

  /**
   * Returns true if this advisor has been closed.
   *
   * @return true if this advisor has been closed
   */
  protected boolean isClosed() {
    return primaryState == CLOSED;
  }

  /**
   * Returns true if this member is currently marked as primary.
   *
   * @return true if this member is currently marked as primary
   */
  public boolean isPrimary() {
    return primaryState == IS_PRIMARY_HOSTING;
  }

  /**
   * Returns true if this member is currently volunteering for primary.
   *
   * @return true if this member is currently volunteering for primary
   */
  private boolean isVolunteering() {
    return primaryState == VOLUNTEERING_HOSTING;
  }

  /**
