/*
 * 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 java.io.DataOutput;
import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.CopyHelper;
import org.apache.geode.DataSerializer;
import org.apache.geode.Delta;
import org.apache.geode.DeltaSerializationException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAlgorithm;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.partition.PartitionListener;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.AtomicLongWithTerminalState;
import org.apache.geode.distributed.internal.DirectReplyProcessor;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.cache.BucketAdvisor.BucketProfile;
import org.apache.geode.internal.cache.CreateRegionProcessor.CreateRegionReplyProcessor;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.event.EventSequenceNumberHolder;
import org.apache.geode.internal.cache.eviction.EvictionController;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.DestroyMessage;
import org.apache.geode.internal.cache.partitioned.InvalidateMessage;
import org.apache.geode.internal.cache.partitioned.LockObject;
import org.apache.geode.internal.cache.partitioned.PRTombstoneMessage;
import org.apache.geode.internal.cache.partitioned.PartitionMessage;
import org.apache.geode.internal.cache.partitioned.PutAllPRMessage;
import org.apache.geode.internal.cache.partitioned.PutMessage;
import org.apache.geode.internal.cache.partitioned.RemoveAllPRMessage;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ClientTombstoneMessage;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessage;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
import org.apache.geode.internal.concurrent.AtomicLong5;
import org.apache.geode.internal.concurrent.Atomics;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;


/**
 * The storage used for a Partitioned Region. This class asserts distributed scope as well as a
 * replicate data policy It does not support transactions
 *
 * Primary election for a BucketRegion can be found in the
 * {@link org.apache.geode.internal.cache.BucketAdvisor} class
 *
 * @since GemFire 5.1
 *
 */
public class BucketRegion extends DistributedRegion implements Bucket {
  private static final Logger logger = LogService.getLogger();

  /**
   * Allows enabling older, buggy behavior where Tx operations might not create
   * VMCachedDeserializable entries.
   */
  private static final boolean TX_PREFERS_NO_SERIALIZED_ENTRIES =
      Boolean.getBoolean("gemfire.tx-prefers-no-serialized-entries");

  @Immutable
  private static final RawValue NULLVALUE = new RawValue(null);
  @Immutable
  public static final RawValue REQUIRES_ENTRY_LOCK = new RawValue(null);
  /**
   * A special value for the bucket size indicating that this bucket has been destroyed.
   */
  private static final long BUCKET_DESTROYED = Long.MIN_VALUE;
  private AtomicLong counter = new AtomicLong();
  private AtomicLong limit;
  private final AtomicLong numOverflowOnDisk = new AtomicLong();
  private final AtomicLong numOverflowBytesOnDisk = new AtomicLong();
  private final AtomicLong numEntriesInVM = new AtomicLong();
  private final AtomicLong evictions = new AtomicLong();
  // For GII
  private CreateRegionReplyProcessor createRegionReplyProcessor;

  /**
   * Contains size in bytes of the values stored in theRealMap. Sizes are tallied during put and
   * remove operations.
   */
  private final AtomicLongWithTerminalState bytesInMemory = new AtomicLongWithTerminalState();

  public static class RawValue {
    private final Object rawValue;

    public RawValue(Object rawVal) {
      rawValue = rawVal;
    }

    public boolean isValueByteArray() {
      return rawValue instanceof byte[];
    }

    public Object getRawValue() {
      return rawValue;
    }

    public void writeAsByteArray(DataOutput out) throws IOException {
      if (isValueByteArray()) {
        DataSerializer.writeByteArray((byte[]) rawValue, out);
      } else if (rawValue instanceof CachedDeserializable) {
        ((CachedDeserializable) rawValue).writeValueAsByteArray(out);
      } else if (Token.isInvalid(rawValue)) {
        DataSerializer.writeByteArray(null, out);
      } else if (rawValue == Token.TOMBSTONE) {
        DataSerializer.writeByteArray(null, out);
      } else {
        DataSerializer.writeObjectAsByteArray(rawValue, out);
      }
    }

    @Override
    public String toString() {
      return "RawValue(" + rawValue + ")";
    }

    /**
     * Return the de-serialized value without changing the stored form in the heap. This causes
     * local access to create a de-serialized copy (extra work) in favor of keeping values in
     * serialized form which is important because it makes remote access more efficient. This
     * assumption is that remote access is much more frequent. TODO Unused, but keeping for
     * potential performance boost when local Bucket access de-serializes the entry (which could
     * hurt perf.)
     *
     * @return the de-serialized value
     */
    public Object getDeserialized(boolean copyOnRead) {
      if (isValueByteArray()) {
        if (copyOnRead) {
          // TODO move this code to CopyHelper.copy?
          byte[] src = (byte[]) rawValue;
          byte[] dest = new byte[src.length];
          System.arraycopy(rawValue, 0, dest, 0, dest.length);
          return dest;
        } else {
          return rawValue;
        }
      } else if (rawValue instanceof CachedDeserializable) {
        if (copyOnRead) {
          return ((CachedDeserializable) rawValue).getDeserializedWritableCopy(null, null);
        } else {
          return ((CachedDeserializable) rawValue).getDeserializedForReading();
        }
      } else if (Token.isInvalid(rawValue)) {
        return null;
      } else {
        if (copyOnRead) {
          return CopyHelper.copy(rawValue);
        } else {
          return rawValue;
        }
      }
    }
  }

  private boolean receivedGatewaySenderStoppedMessage = false;

  private final int redundancy;

  /** the partitioned region to which this bucket belongs */
  private final PartitionedRegion partitionedRegion;
  private final Map<Object, ExpiryTask> pendingSecondaryExpires = new HashMap<>();

  /* one map per bucket region */
  private final HashMap<Object, LockObject> allKeysMap = new HashMap<>();

  static final boolean FORCE_LOCAL_LISTENERS_INVOCATION = Boolean
      .getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "BucketRegion.alwaysFireLocalListeners");

  private volatile AtomicLong5 eventSeqNum = null;

  AtomicLong5 getEventSeqNum() {
    return eventSeqNum;
  }

  public BucketRegion(String regionName, RegionAttributes attrs, LocalRegion parentRegion,
      InternalCache cache, InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock) {
    super(regionName, attrs, parentRegion, cache, internalRegionArgs, statisticsClock);
    if (PartitionedRegion.DISABLE_SECONDARY_BUCKET_ACK) {
      Assert.assertTrue(attrs.getScope().isDistributedNoAck());
    } else {
      Assert.assertTrue(attrs.getScope().isDistributedAck());
    }
    Assert.assertTrue(attrs.getDataPolicy().withReplication());
    Assert.assertTrue(!attrs.getEarlyAck());
    Assert.assertTrue(isUsedForPartitionedRegionBucket());
    Assert.assertTrue(!isUsedForPartitionedRegionAdmin());
    Assert.assertTrue(internalRegionArgs.getBucketAdvisor() != null);
    Assert.assertTrue(internalRegionArgs.getPartitionedRegion() != null);
    redundancy = internalRegionArgs.getPartitionedRegionBucketRedundancy();
    partitionedRegion = internalRegionArgs.getPartitionedRegion();
    setEventSeqNum();
  }

  // Attempt to direct the GII process to the primary first
  @Override
  public void initialize(InputStream snapshotInputStream, InternalDistributedMember imageTarget,
      InternalRegionArguments internalRegionArgs)
      throws TimeoutException, IOException, ClassNotFoundException {
    // Set this region in the ProxyBucketRegion early so that profile exchange will
    // perform the correct fillInProfile method
    getBucketAdvisor().getProxyBucketRegion().setBucketRegion(this);
    boolean success = false;
    try {
      final InternalDistributedMember primaryHolder = getBucketAdvisor().basicGetPrimaryMember();
      if (primaryHolder != null && !primaryHolder.equals(getMyId())) {
        // Ignore the provided image target, use an existing primary (if any)
        super.initialize(snapshotInputStream, primaryHolder, internalRegionArgs);
      } else {
        super.initialize(snapshotInputStream, imageTarget, internalRegionArgs);
      }

      success = true;
    } finally {
      if (!success) {
        removeFromPeersAdvisors(false);
        getBucketAdvisor().getProxyBucketRegion().clearBucketRegion(this);
      }
    }
  }

  private void setEventSeqNum() {
    if (partitionedRegion.isShadowPR() && partitionedRegion.getColocatedWith() != null) {
      PartitionedRegion parentPR = ColocationHelper.getLeaderRegion(partitionedRegion);
      BucketRegion parentBucket = parentPR.getDataStore().getLocalBucketById(getId());
      // needs to be set only once.
      if (parentBucket.eventSeqNum == null) {
        parentBucket.eventSeqNum = new AtomicLong5(getId());
      }
    }
    if (partitionedRegion.getColocatedWith() == null) {
      eventSeqNum = new AtomicLong5(getId());
    } else {
      PartitionedRegion parentPR = ColocationHelper.getLeaderRegion(partitionedRegion);
      BucketRegion parentBucket = parentPR.getDataStore().getLocalBucketById(getId());
      if (parentBucket == null && logger.isDebugEnabled()) {
        logger.debug("The parentBucket of region {} bucketId {} is NULL",
            partitionedRegion.getFullPath(), getId());
      }
      Assert.assertTrue(parentBucket != null);
      eventSeqNum = parentBucket.eventSeqNum;
    }
  }


  @Override
  void initialized() {
    // announce that the bucket is ready
    // setHosting performs a profile exchange, so there
    // is no need to call super.initialized() here.
  }

  @Override
  DiskStoreImpl findDiskStore(RegionAttributes regionAttributes,
      InternalRegionArguments internalRegionArgs) {
    return internalRegionArgs.getPartitionedRegion().getDiskStore();
  }

  @Override
  public void registerCreateRegionReplyProcessor(CreateRegionReplyProcessor processor) {
    createRegionReplyProcessor = processor;
  }

  @Override
  protected void recordEventStateFromImageProvider(InternalDistributedMember provider) {
    if (createRegionReplyProcessor != null) {
      Map<ThreadIdentifier, EventSequenceNumberHolder> providerEventStates =
          createRegionReplyProcessor.getEventState(provider);
      if (providerEventStates != null) {
        recordEventState(provider, providerEventStates);
      } else {
        // Does not see this to happen. Just in case we get gii from a node
        // that was not in the cluster originally when we sent
        // createRegionMessage (its event tracker was saved),
        // but later available before we could get gii from anyone else.
        // This will not cause data inconsistent issue. Log this message for debug purpose.
        logger.info("Could not initiate event tracker from GII provider {}", provider);
      }
      createRegionReplyProcessor = null;
    }
  }

  @Override
  protected CacheDistributionAdvisor createDistributionAdvisor(
      InternalRegionArguments internalRegionArgs) {
    return internalRegionArgs.getBucketAdvisor();
  }

  @Override
  public BucketAdvisor getBucketAdvisor() {
    return (BucketAdvisor) getDistributionAdvisor();
  }

  @Override
  public boolean isHosting() {
    return getBucketAdvisor().isHosting();
  }

  @Override
  protected EventID distributeTombstoneGC(Set<Object> keysRemoved) {
    EventID eventId = super.distributeTombstoneGC(keysRemoved);
    if (keysRemoved != null && keysRemoved.size() > 0 && getFilterProfile() != null) {
      // send the GC to members that don't have the bucket but have the PR so they
      // can forward the event to clients
      PRTombstoneMessage.send(this, keysRemoved, eventId);
    }
    return eventId;
  }

  @Override
  boolean needsTombstoneGCKeysForClients(EventID eventID, FilterInfo clientRouting) {
    if (eventID == null) {
      return false;
    }
    if (CacheClientNotifier.getInstance() == null) {
      return false;
    }
    if (clientRouting != null) {
      return true;
    }
    return getFilterProfile() != null;
  }

  @Override
  void notifyClientsOfTombstoneGC(Map<VersionSource<?>, Long> regionGCVersions,
      Set<Object> removedKeys, EventID eventID, FilterInfo routing) {
    if (CacheClientNotifier.singletonHasClientProxies()) {
      // Only route the event to clients interested in the partitioned region.
      // We do this by constructing a region-level event and then use it to
      // have the filter profile ferret out all of the clients that have interest
      // in this region
      FilterProfile fp = getFilterProfile();
      // fix for bug #46309 - don't send null/empty key set to clients
      if ((removedKeys != null && !removedKeys.isEmpty()) // bug #51877 - NPE in clients
          && (routing != null || (fp != null && fp.hasInterest()))) {
        RegionEventImpl regionEvent = new RegionEventImpl(getPartitionedRegion(),
            Operation.REGION_DESTROY, null, true, getMyId());
        FilterInfo clientRouting = routing;
        if (clientRouting == null) {
          clientRouting = fp.getLocalFilterRouting(regionEvent);
        }
        regionEvent.setLocalFilterInfo(clientRouting);

        ClientUpdateMessage clientMessage =
            ClientTombstoneMessage.gc(getPartitionedRegion(), removedKeys, eventID);
        CacheClientNotifier.notifyClients(regionEvent, clientMessage);
      }
    }
  }

  /**
   * Search the CM for keys. If found any, return the first found one Otherwise, save the keys into
   * the CM, and return null The thread will acquire the lock before searching.
   *
   * @return first key found in CM null means not found
   */
  LockObject searchAndLock(Object[] keys) {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    LockObject foundLock = null;

    synchronized (allKeysMap) {
      // check if there's any key in map
      for (Object key : keys) {
        if (allKeysMap.containsKey(key)) {
          foundLock = allKeysMap.get(key);
          if (isDebugEnabled) {
            logger.debug("LockKeys: found key: {}:{}", key, foundLock.lockedTimeStamp);
          }
          foundLock.waiting();
          break;
        }
      }

      // save the keys when still locked
      if (foundLock == null) {
        for (Object key : keys) {
          LockObject lockValue =
              new LockObject(key, isDebugEnabled ? System.currentTimeMillis() : 0);
          allKeysMap.put(key, lockValue);
          if (isDebugEnabled) {
            logger.debug("LockKeys: add key: {}:{}", key, lockValue.lockedTimeStamp);
          }
        }
      }
    }

    return foundLock;
  }

  /**
   * After processed the keys, this method will remove them from CM. And notifyAll for each key. The
   * thread needs to acquire lock of CM first.
   */
  public void removeAndNotifyKeys(Object[] keys) {
    final boolean isTraceEnabled = logger.isTraceEnabled();

    synchronized (allKeysMap) {
      for (Object key : keys) {
        LockObject lockValue = allKeysMap.remove(key);
        if (lockValue != null) {
          // let current thread become the monitor of the key object
          synchronized (lockValue) {
            lockValue.setRemoved();
            if (isTraceEnabled) {
              long waitTime = System.currentTimeMillis() - lockValue.lockedTimeStamp;
              logger.trace("LockKeys: remove key {}, notifyAll for {}. It waited {}", key,
                  lockValue, waitTime);
            }
            if (lockValue.isSomeoneWaiting()) {
              lockValue.notifyAll();
            }
          }
        }
      } // for
    }
  }

  /**
   * Keep checking if CM has contained any key in keys. If yes, wait for notify, then retry again.
   * This method will block current thread for long time. It only exits when current thread
   * successfully save its keys into CM.
   */
  public boolean waitUntilLocked(Object[] keys) {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    final String title = "BucketRegion.waitUntilLocked:";
    while (true) {
      LockObject foundLock = searchAndLock(keys);

      if (foundLock != null) {
        synchronized (foundLock) {
          try {
            while (!foundLock.isRemoved()) {
              partitionedRegion.checkReadiness();
              foundLock.wait(1000);
              // primary could be changed by prRebalancing while waiting here
              checkForPrimary();
            }
          } catch (InterruptedException e) {
            // TODO this isn't a localizable string and it's being logged at info level
            if (isDebugEnabled) {
              logger.debug("{} interrupted while waiting for {}", title, foundLock);
            }
          }
          if (isDebugEnabled) {
            long waitTime = System.currentTimeMillis() - foundLock.lockedTimeStamp;
            logger.debug("{} waited {} ms to lock {}", title, waitTime, foundLock);
          }
        }
      } else {
        // now the keys have been locked by this thread
        return true;
      } // to lock and process
    } // while
  }

  // Entry (Put/Create) rules
  // If this is a primary for the bucket
  // 1) apply op locally, aka update or create entry
  // 2) distribute op to bucket secondaries and cache servers with synchrony on local entry
  // 3) cache listener with synchrony on entry
  // Else not a primary
  // 1) apply op locally
  // 2) update local bs, gateway
  @Override
  public boolean virtualPut(EntryEventImpl event, boolean ifNew, boolean ifOld,
      Object expectedOldValue, boolean requireOldValue, long lastModified,
      boolean overwriteDestroyed, boolean invokeCallbacks,
      boolean throwConcurrentModification)
      throws TimeoutException, CacheWriterException {

    boolean isLocked = lockKeysAndPrimary(event);

    try {
      if (partitionedRegion.isParallelWanEnabled()) {
        handleWANEvent(event);
      }
      if (!hasSeenEvent(event)) {
        forceSerialized(event);
        RegionEntry oldEntry = entries.basicPut(event, lastModified, ifNew, ifOld,
            expectedOldValue, requireOldValue, overwriteDestroyed);
        return oldEntry != null;
      }
      if (event.getDeltaBytes() != null && event.getRawNewValue() == null) {
        // This means that this event has delta bytes but no full value.
        // Request the full value of this event.
        // The value in this vm may not be same as this event's value.
        throw new InvalidDeltaException(
            "Cache encountered replay of event containing delta bytes for key " + event.getKey());
      }
      // Forward the operation and event messages
      // to members with bucket copies that may not have seen the event. Their
      // EventTrackers will keep them from applying the event a second time if
      // they've already seen it.
      if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
        logger.trace(LogMarker.DM_VERBOSE,
            "BR.virtualPut: this cache has already seen this event {}", event);
      }
      if (!getConcurrencyChecksEnabled() || event.hasValidVersionTag()) {
        distributeUpdateOperation(event, lastModified);
      }
      return true;
    } finally {
      if (isLocked) {
        releaseLockForKeysAndPrimary(event);
      }
    }
  }


  long generateTailKey() {
    long key = eventSeqNum.addAndGet(partitionedRegion.getTotalNumberOfBuckets());
    if (key < 0 || key % getPartitionedRegion().getTotalNumberOfBuckets() != getId()) {
      logger.error("ERROR! The sequence number {} generated for the bucket {} is incorrect.",
          new Object[] {key, getId()});
    }
    if (logger.isDebugEnabled()) {
      logger.debug("WAN: On primary bucket {}, setting the seq number as {}", getId(),
          eventSeqNum.get());
    }
    return eventSeqNum.get();
  }

  @Override
  public void txHandleWANEvent(EntryEventImpl event) {
    if (partitionedRegion.isParallelWanEnabled()) {
      handleWANEvent(event);
    }
  }

  @Override
  public void handleWANEvent(EntryEventImpl event) {
    if (eventSeqNum == null) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "The bucket corresponding to this user bucket is not created yet. This event will not go to remote wan site. Event: {}",
            event);
      }
    }

    if (!(this instanceof AbstractBucketRegionQueue)) {
      if (getBucketAdvisor().isPrimary()) {
        long key = eventSeqNum.addAndGet(partitionedRegion.getTotalNumberOfBuckets());
        if (key < 0 || key % getPartitionedRegion().getTotalNumberOfBuckets() != getId()) {
          logger.error("ERROR! The sequence number {} generated for the bucket {} is incorrect.",
              new Object[] {key, getId()});
        }
        event.setTailKey(key);
        if (logger.isDebugEnabled()) {
          logger.debug("WAN: On primary bucket {}, setting the seq number as {}", getId(),
              eventSeqNum.get());
        }
      } else {
        // Can there be a race here? Like one thread has done put in primary but
        // its update comes later
        // in that case its possible that a tail key is missed.
        // we can handle that by only incrementing the tailKey and never
        // setting it less than the current value.
        Atomics.setIfGreater(eventSeqNum, event.getTailKey());
        if (logger.isDebugEnabled()) {
          logger.debug("WAN: On secondary bucket {}, setting the seq number as {}", getId(),
              event.getTailKey());
        }
      }
    }
  }

  /**
   * Fix for Bug#45917 We are updating the seqNumber so that new seqNumbers are generated starting
   * from the latest in the system.
   */
  void updateEventSeqNum(long l) {
    Atomics.setIfGreater(eventSeqNum, l);
    if (logger.isDebugEnabled()) {
      logger.debug("WAN: On bucket {}, setting the seq number as {} before GII", getId(), l);
    }
  }

  protected void distributeUpdateOperation(EntryEventImpl event, long lastModified) {
    long token = -1;
    UpdateOperation op = null;

    try {
      if (!event.isOriginRemote() && !event.isNetSearch() && getBucketAdvisor().isPrimary()) {
        if (event.isBulkOpInProgress()) {
          // consolidate the UpdateOperation for each entry into a PutAllMessage
          // since we did not call basicPutPart3(), so we have to explicitly addEntry here
          event.getPutAllOperation().addEntry(event, getId());
        } else {
          // before distribute: BR's put
          op = new UpdateOperation(event, lastModified);
          token = op.startOperation();
          if (logger.isDebugEnabled()) {
            logger.debug("sent update operation : for region  : {}: with event: {}", getName(),
                event);
          }
        }
      }
      if (!event.getOperation().isPutAll()) { // putAll will invoke listeners later
        event.invokeCallbacks(this, true, true);
      }
    } finally {
      if (op != null) {
        op.endOperation(token);
      }
    }
  }

  /**
   * distribute the operation in basicPutPart2 so the region entry lock is held
   */
  @Override
  public long basicPutPart2(EntryEventImpl event, RegionEntry entry, boolean isInitialized,
      long lastModified, boolean clearConflict) {
    // Assumed this is called with entry synchrony

    // Typically UpdateOperation is called with the
    // timestamp returned from basicPutPart2, but as a bucket we want to do
    // distribution *before* we do basicPutPart2.
    final long modifiedTime = event.getEventTime(lastModified);

    long token = -1;
    UpdateOperation op = null;

    try {
      // Update the get stats if necessary.
      if (partitionedRegion.getDataStore().hasClientInterest(event)) {
        updateStatsForGet(entry, true);
      }
      if (!event.isOriginRemote()) {
        if (event.getVersionTag() == null || event.getVersionTag().isGatewayTag()) {
          boolean eventHasDelta = event.getDeltaBytes() != null;
          VersionTag v = entry.generateVersionTag(null, eventHasDelta, this, event);
          if (v != null) {
            if (logger.isDebugEnabled()) {
              logger.debug("generated version tag {} in region {}", v, getName());
            }
          }
        }

        // This code assumes it is safe ignore token mode (GII in progress)
        // because it assumes when the origin of the event is local,
        // the GII has completed and the region is initialized and open for local
        // ops

        if (!event.isBulkOpInProgress()) {
          long start = partitionedRegion.getPrStats().startSendReplication();
          try {
            // before distribute: PR's put PR
            op = new UpdateOperation(event, modifiedTime);
            token = op.startOperation();
          } finally {
            partitionedRegion.getPrStats().endSendReplication(start);
          }
        } else {
          // consolidate the UpdateOperation for each entry into a PutAllMessage
          // basicPutPart3 takes care of this
        }
      }

      return super.basicPutPart2(event, entry, isInitialized, lastModified, clearConflict);
    } finally {
      if (op != null) {
        op.endOperation(token);
      }
    }
  }

  @Override
  protected void notifyGatewaySender(EnumListenerEvent operation, EntryEventImpl event) {
    // We don't need to clone the event for new Gateway Senders.
    // Preserve the bucket reference for resetting it later.
    InternalRegion bucketRegion = event.getRegion();
    try {
      event.setRegion(partitionedRegion);
      partitionedRegion.notifyGatewaySender(operation, event);
    } finally {
      // reset the event region back to bucket region.
      // This should work as gateway queue create GatewaySenderEvent for
      // queueing.
      event.setRegion(bucketRegion);
    }
  }

  public void checkForPrimary() {
    final boolean isp = getBucketAdvisor().isPrimary();
    if (!isp) {
      partitionedRegion.checkReadiness();
      checkReadiness();
      InternalDistributedMember primaryHolder = getBucketAdvisor().basicGetPrimaryMember();
      throw new PrimaryBucketException(
          "Bucket " + getName() + " is not primary. Current primary holder is " + primaryHolder);
    }
  }

  /**
   * Checks to make sure that this node is primary, and locks the bucket to make sure the bucket
   * stays the primary bucket while the write is in progress. This method must be followed with
   * a call to releaseLockForKeysAndPrimary() if keys and primary are locked.
   */
  boolean lockKeysAndPrimary(EntryEventImpl event) {
    if (!needWriteLock(event)) {
      return false;
    }

    if (cache.isCacheAtShutdownAll()) {
      throw cache.getCacheClosedException("Cache is shutting down");
    }

    Object[] keys = getKeysToBeLocked(event);
    waitUntilLocked(keys); // it might wait for long time

    if (wasPrimaryLockedPreviously(event)) {
      return true;
    }
    boolean lockedForPrimary = false;

    try {
      lockedForPrimary = doLockForPrimary(false);
      // tryLock is false means doLockForPrimary won't return false.
      // either the method returns true or fails with an exception
      assert lockedForPrimary : "expected doLockForPrimary returns true";
      return lockedForPrimary;
    } finally {
      if (!lockedForPrimary) {
        removeAndNotifyKeys(keys);
      }
    }
  }

  Object[] getKeysToBeLocked(EntryEventImpl event) {
    Object[] keys = new Object[1];
    keys[0] = event.getKey();
    return keys;
  }

  /**
   * lock this bucket
   *
   * @param tryLock - whether to use tryLock (true) or a blocking lock (false)
   * @return true if locks were obtained and are still held
   */
  public boolean doLockForPrimary(boolean tryLock) {
    boolean locked = lockPrimaryStateReadLock(tryLock);
    if (!locked) {
      return false;
    }

    boolean isPrimary = false;
    try {
      // Throw a PrimaryBucketException if this VM is assumed to be the
      // primary but isn't, preventing update and distribution
      checkForPrimary();

      if (cache.isCacheAtShutdownAll()) {
        throw cache.getCacheClosedException("Cache is shutting down");
      }

      isPrimary = true;
    } finally {
      if (!isPrimary) {
        doUnlockForPrimary();
      }
    }

    return true;
  }

  private boolean lockPrimaryStateReadLock(boolean tryLock) {
    Lock primaryMoveReadLock = getBucketAdvisor().getPrimaryMoveReadLock();
    for (;;) {
      boolean interrupted = Thread.interrupted();
      try {
        // Get the lock. If we have to wait here, it's because
        // this VM is actively becoming "not primary". We don't want
        // to throw an exception until this VM is actually no longer
        // primary, so we wait here for not primary to complete.
        if (tryLock) {
          boolean locked = primaryMoveReadLock.tryLock();
          if (!locked) {
            return false;
          }
        } else {
          primaryMoveReadLock.lockInterruptibly();
