geode-core/src/main/java/org/apache/geode/internal/cache/AbstractRegionMap.java - geode - Git at Google

 /*
  * 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.util.Collection;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.GemFireIOException;
 import org.apache.geode.InvalidDeltaException;
 import org.apache.geode.annotations.internal.MutableForTesting;
 import org.apache.geode.cache.CacheWriterException;
 import org.apache.geode.cache.DiskAccessException;
 import org.apache.geode.cache.EntryNotFoundException;
 import org.apache.geode.cache.Operation;
 import org.apache.geode.cache.RegionDestroyedException;
 import org.apache.geode.cache.TimeoutException;
 import org.apache.geode.cache.TransactionId;
 import org.apache.geode.cache.query.IndexMaintenanceException;
 import org.apache.geode.cache.query.QueryException;
 import org.apache.geode.cache.query.internal.index.IndexManager;
 import org.apache.geode.cache.query.internal.index.IndexProtocol;
 import org.apache.geode.distributed.internal.DistributionManager;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.Assert;
 import org.apache.geode.internal.cache.DiskInitFile.DiskRegionFlag;
 import org.apache.geode.internal.cache.entries.AbstractOplogDiskRegionEntry;
 import org.apache.geode.internal.cache.entries.AbstractRegionEntry;
 import org.apache.geode.internal.cache.entries.DiskEntry;
 import org.apache.geode.internal.cache.entries.OffHeapRegionEntry;
 import org.apache.geode.internal.cache.map.CacheModificationLock;
 import org.apache.geode.internal.cache.map.FocusedRegionMap;
 import org.apache.geode.internal.cache.map.RegionMapCommitPut;
 import org.apache.geode.internal.cache.map.RegionMapDestroy;
 import org.apache.geode.internal.cache.map.RegionMapPut;
 import org.apache.geode.internal.cache.persistence.DiskRegionView;
 import org.apache.geode.internal.cache.region.entry.RegionEntryFactoryBuilder;
 import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
 import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
 import org.apache.geode.internal.cache.versions.RegionVersionVector;
 import org.apache.geode.internal.cache.versions.VersionHolder;
 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.logging.log4j.LogMarker;
 import org.apache.geode.internal.offheap.OffHeapHelper;
 import org.apache.geode.internal.offheap.OffHeapRegionEntryHelper;
 import org.apache.geode.internal.offheap.StoredObject;
 import org.apache.geode.internal.offheap.annotations.Released;
 import org.apache.geode.internal.offheap.annotations.Retained;
 import org.apache.geode.internal.sequencelog.EntryLogger;
 import org.apache.geode.internal.size.ReflectionSingleObjectSizer;
 import org.apache.geode.internal.util.concurrent.ConcurrentMapWithReusableEntries;
 import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap;
 import org.apache.geode.logging.internal.log4j.api.LogService;

 /**
  * Abstract implementation of {@link RegionMap}that has all the common behavior.
  *
  * @since GemFire 3.5.1
  */
 public abstract class AbstractRegionMap extends BaseRegionMap
     implements FocusedRegionMap, CacheModificationLock {
   private static final Logger logger = LogService.getLogger();
   private final TxCallbackEventFactory txCallbackEventFactory = new TxCallbackEventFactoryImpl();

   /** The underlying map for this region. */
   protected ConcurrentMapWithReusableEntries<Object, Object> map;

   /**
    * This test hook is used to force the conditions during entry destroy. This hook is used by
    * DestroyEntryWithConcurrentOperationJUnitTest.
    */
   @MutableForTesting
   static final Runnable testHookRunnableForConcurrentOperation = null;

   private RegionEntryFactory entryFactory;

   private Attributes attr;

   // the region that owns this map
   private RegionMapOwner owner;

   private final EntryEventSerialization entryEventSerialization = new EntryEventSerialization();

   protected AbstractRegionMap(InternalRegionArguments internalRegionArgs) {
     // do nothing
   }

   protected void initialize(RegionMapOwner owner, Attributes attr,
       InternalRegionArguments internalRegionArgs, boolean isLRU) {
     _setAttributes(attr);
     setOwner(owner);
     setEntryMap(createConcurrentMapWithReusableEntries(attr.initialCapacity, attr.loadFactor,
         attr.concurrencyLevel, false, new AbstractRegionEntry.HashRegionEntryCreator()));

     boolean isDisk;
     boolean withVersioning;
     boolean offHeap;
     if (owner instanceof InternalRegion) {
       InternalRegion region = (InternalRegion) owner;
       isDisk = region.getDiskRegion() != null;
       withVersioning = region.getConcurrencyChecksEnabled();
       offHeap = region.getOffHeap();
     } else if (owner instanceof PlaceHolderDiskRegion) {
       offHeap = ((RegionEntryContext) owner).getOffHeap();
       isDisk = true;
       withVersioning =
           ((DiskRegionView) owner).getFlags().contains(DiskRegionFlag.IS_WITH_VERSIONING);
     } else {
       throw new IllegalStateException("expected LocalRegion or PlaceHolderDiskRegion");
     }

     setEntryFactory(new RegionEntryFactoryBuilder().create(attr.statisticsEnabled, isLRU, isDisk,
         withVersioning, offHeap));
   }

   private ConcurrentMapWithReusableEntries<Object, Object> createConcurrentMapWithReusableEntries(
       int initialCapacity, float loadFactor, int concurrencyLevel, boolean isIdentityMap,
       CustomEntryConcurrentHashMap.HashEntryCreator<Object, Object> entryCreator) {
     if (entryCreator != null) {
       return new CustomEntryConcurrentHashMap<>(initialCapacity, loadFactor, concurrencyLevel,
           isIdentityMap, entryCreator);
     } else {
       return new CustomEntryConcurrentHashMap<>(initialCapacity, loadFactor, concurrencyLevel,
           isIdentityMap);
     }
   }

   @Override
   public void changeOwner(LocalRegion r) {
     if (r == _getOwnerObject()) {
       return;
     }
     setOwner(r);
   }

   @Override
   public void setEntryFactory(RegionEntryFactory f) {
     this.entryFactory = f;
   }

   @Override
   public RegionEntryFactory getEntryFactory() {
     return this.entryFactory;
   }

   private void _setAttributes(Attributes a) {
     this.attr = a;
   }

   @Override
   public Attributes getAttributes() {
     return this.attr;
   }

   public LocalRegion _getOwner() {
     return (LocalRegion) this.owner;
   }

   boolean _isOwnerALocalRegion() {
     return this.owner instanceof LocalRegion;
   }

   Object _getOwnerObject() {
     return this.owner;
   }

   /**
    * Tells this map what region owns it.
    */
   private void setOwner(RegionMapOwner owner) {
     this.owner = owner;
   }

   @Override
   public ConcurrentMapWithReusableEntries<Object, Object> getCustomEntryConcurrentHashMap() {
     return map;
   }

   @Override
   public Map<Object, Object> getEntryMap() {
     return map;
   }

   @Override
   public void setEntryMap(ConcurrentMapWithReusableEntries<Object, Object> map) {
     this.map = map;
   }

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

   // this is currently used by stats and eviction
   @Override
   public int sizeInVM() {
     return getEntryMap().size();
   }

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

   @Override
   public Set keySet() {
     return getEntryMap().keySet();
   }

   @Override
   @SuppressWarnings({"unchecked", "rawtypes"})
   public Collection<RegionEntry> regionEntries() {
     return (Collection) getEntryMap().values();
   }

   @SuppressWarnings({"unchecked", "rawtypes"})
   @Override
   public Collection<RegionEntry> regionEntriesInVM() {
     return (Collection) getEntryMap().values();
   }

   @Override
   public boolean containsKey(Object key) {
     RegionEntry re = getEntry(key);
     if (re == null) {
       return false;
     }
     if (re.isRemoved()) {
       return false;
     }
     return true;
   }

   @Override
   public RegionEntry getEntry(Object key) {
     RegionEntry re = (RegionEntry) getEntryMap().get(key);
     return re;
   }

   @Override
   public RegionEntry getEntry(EntryEventImpl event) {
     return getEntry(event.getKey());
   }


   @Override
   public RegionEntry getEntryInVM(Object key) {
     return (RegionEntry) getEntryMap().get(key);
   }

   @Override
   public RegionEntry putEntryIfAbsent(Object key, RegionEntry regionEntry) {
     RegionEntry oldRe = (RegionEntry) getEntryMap().putIfAbsent(key, regionEntry);
     if (oldRe == null && (regionEntry instanceof OffHeapRegionEntry) && _isOwnerALocalRegion()
         && _getOwner().isThisRegionBeingClosedOrDestroyed()) {
       // prevent orphan during concurrent destroy (#48068)
       Object v = regionEntry.getValue();
       if (v != Token.REMOVED_PHASE1 && v != Token.REMOVED_PHASE2 && v instanceof StoredObject
           && ((StoredObject) v).hasRefCount()) {
         if (getEntryMap().remove(key, regionEntry)) {
           ((OffHeapRegionEntry) regionEntry).release();
         }
       }
     }
     return oldRe;
   }

   @Override
   public RegionEntry getOperationalEntryInVM(Object key) {
     RegionEntry re = (RegionEntry) getEntryMap().get(key);
     return re;
   }


   @Override
   public void removeEntry(Object key, RegionEntry regionEntry, boolean updateStat) {
     if (regionEntry.isTombstone() && getEntryMap().get(key) == regionEntry) {
       logger.fatal(
           "Internal product error: attempt to directly remove a versioned tombstone from region entry map",
           new Exception("stack trace"));
       return; // can't remove tombstones except from the tombstone sweeper
     }
     if (getEntryMap().remove(key, regionEntry)) {
       regionEntry.removePhase2();
       if (updateStat) {
         incEntryCount(-1);
       }
     }
   }

   @Override
   public void removeEntry(Object key, RegionEntry regionEntry, boolean updateStat,
       EntryEventImpl event, final InternalRegion internalRegion) {
     boolean success = false;
     if (regionEntry.isTombstone() && getEntryMap().get(key) == regionEntry) {
       logger.fatal(
           "Internal product error: attempt to directly remove a versioned tombstone from region entry map",
           new Exception("stack trace"));
       return; // can't remove tombstones except from the tombstone sweeper
     }
     if (getEntryMap().remove(key, regionEntry)) {
       regionEntry.removePhase2();
       success = true;
       if (updateStat) {
         incEntryCount(-1);
       }
     }
   }

   @Override
   public void incEntryCount(int delta) {
     LocalRegion lr = _getOwner();
     if (lr != null) {
       CachePerfStats stats = lr.getCachePerfStats();
       if (stats != null) {
         stats.incEntryCount(delta);
       }
     }
   }

   void incClearCount(LocalRegion lr) {
     if (lr != null && !(lr instanceof HARegion)) {
       CachePerfStats stats = lr.getCachePerfStats();
       if (stats != null) {
         stats.incClearCount();
       }
     }
   }

   private void _mapClear() {
     Executor executor = null;
     InternalCache cache = this.owner.getCache();
     if (cache != null) {
       DistributionManager manager = cache.getDistributionManager();
       if (manager != null) {
         executor = manager.getExecutors().getWaitingThreadPool();
       }
     }
     getCustomEntryConcurrentHashMap().clearWithExecutor(executor);
   }

   @Override
   public void close(BucketRegion bucketRegion) {
     clear(null, bucketRegion);
   }

   /**
    * Clear the region and, if an RVV is given, return a collection of the version sources in all
    * remaining tags
    */
   @Override
   public Set<VersionSource> clear(RegionVersionVector rvv, BucketRegion bucketRegion) {
     Set<VersionSource> result = new HashSet<VersionSource>();

     if (!_isOwnerALocalRegion()) {
       // Fix for #41333. Just clear the the map
       // if we failed during initialization.
       _mapClear();
       return null;
     }
     if (logger.isDebugEnabled()) {
       logger.debug("Clearing entries for {} rvv={}", _getOwner(), rvv);
     }
     LocalRegion lr = _getOwner();
     RegionVersionVector localRvv = lr.getVersionVector();
     incClearCount(lr);
     // lock for size calcs if the region might have tombstones
     Object lockObj = lr.getConcurrencyChecksEnabled() ? lr.getSizeGuard() : new Object();
     synchronized (lockObj) {
       if (rvv == null) {
         int delta = 0;
         try {
           delta = sizeInVM(); // TODO soplog need to determine if stats should
                               // reflect only size in memory or the complete thing
         } catch (GemFireIOException e) {
           // ignore rather than throwing an exception during cache close
         }
         int tombstones = lr.getTombstoneCount();
         _mapClear();
         _getOwner().updateSizeOnClearRegion(delta - tombstones);
         _getOwner().incTombstoneCount(-tombstones);
         if (delta != 0) {
           incEntryCount(-delta);
         }
       } else {
         int delta = 0;
         int tombstones = 0;
         VersionSource myId = _getOwner().getVersionMember();
         if (localRvv != rvv) {
           localRvv.recordGCVersions(rvv);
         }
         final boolean isTraceEnabled = logger.isTraceEnabled();
         for (RegionEntry re : regionEntries()) {
           synchronized (re) {
             Token value = re.getValueAsToken();
             // if it's already being removed or the entry is being created we leave it alone
             if (value == Token.REMOVED_PHASE1 || value == Token.REMOVED_PHASE2) {
               continue;
             }

             VersionSource id = re.getVersionStamp().getMemberID();
             if (id == null) {
               id = myId;
             }
             if (rvv.contains(id, re.getVersionStamp().getRegionVersion())) {
               if (isTraceEnabled) {
                 logger.trace("region clear op is removing {} {}", re.getKey(),
                     re.getVersionStamp());
               }

               boolean tombstone = re.isTombstone();
               // note: it.remove() did not reliably remove the entry so we use remove(K,V) here
               if (getEntryMap().remove(re.getKey(), re)) {
                 if (OffHeapRegionEntryHelper.doesClearNeedToCheckForOffHeap()) {
                   GatewaySenderEventImpl.release(re.getValue()); // OFFHEAP _getValue ok
                 }
                 // If this is an overflow only region, we need to free the entry on
                 // disk at this point.
                 try {
                   re.removePhase1(lr, true);
                 } catch (RegionClearedException e) {
                   // do nothing, it's already cleared.
                 }
                 re.removePhase2();
                 lruEntryDestroy(re);
                 if (tombstone) {
                   _getOwner().incTombstoneCount(-1);
                   tombstones += 1;
                 } else {
                   delta += 1;
                 }
               }
             } else { // rvv does not contain this entry so it is retained
               result.add(id);
             }
           }
         }
         _getOwner().updateSizeOnClearRegion(delta);
         incEntryCount(-delta);
         incEntryCount(-tombstones);
         if (logger.isDebugEnabled()) {
           logger.debug("Size after clearing = {}", getEntryMap().size());
         }
         if (isTraceEnabled && getEntryMap().size() < 20) {
           _getOwner().dumpBackingMap();
         }
       }
     }
     return result;
   }

   public void lruUpdateCallback(boolean b) {
     // By default do nothing; LRU maps needs to override this method
   }

   /**
    * Tell an LRU that a new entry has been created
    */
   @Override
   public void lruEntryCreate(RegionEntry e) {
     // do nothing by default
   }

   /**
    * Tell an LRU that an existing entry has been destroyed
    */
   @Override
   public void lruEntryDestroy(RegionEntry regionEntry) {
     // do nothing by default
   }

   /**
    * Tell an LRU that an existing entry has been modified
    */
   @Override
   public void lruEntryUpdate(RegionEntry e) {
     // do nothing by default
   }

   @Override
   public void decTxRefCount(RegionEntry e) {
     LocalRegion lr = null;
     if (_isOwnerALocalRegion()) {
       lr = _getOwner();
     }
     e.decRefCount(null, lr);
   }

   /**
    * Process an incoming version tag for concurrent operation detection. This must be done before
    * modifying the region entry.
    *
    * @param regionEntry the entry that is to be modified
    * @param event the modification to the entry
    * @throws InvalidDeltaException if the event contains a delta that cannot be applied
    * @throws ConcurrentCacheModificationException if the event is in conflict with a previously
    *         applied change
    */
   @Override
   public void processVersionTag(RegionEntry regionEntry, EntryEventImpl event) {
     if (regionEntry.getVersionStamp() != null) {
       regionEntry.getVersionStamp().processVersionTag(event);

       // during initialization we record version tag info to detect ops the
       // image provider hasn't seen
       VersionTag<?> tag = event.getVersionTag();
       if (tag != null && !event.getRegion().isInitialized()) {
         ImageState is = event.getRegion().getImageState();
         if (is != null && !event.getRegion().isUsedForPartitionedRegionBucket()) {
           if (logger.isTraceEnabled()) {
             logger.trace("recording version tag in image state: {}", tag);
           }
           is.addVersionTag(event.getKey(), tag);
         }
       }
     }
   }

   private void processVersionTagForGII(RegionEntry re, LocalRegion owner, VersionTag entryVersion,
       boolean isTombstone, InternalDistributedMember sender, boolean checkConflicts) {

     re.getVersionStamp().processVersionTag(_getOwner(), entryVersion, isTombstone, false,
         owner.getMyId(), sender, checkConflicts);
   }

   @Override
   public void copyRecoveredEntries(RegionMap rm) {
     // We need to sort the tombstones before scheduling them,
     // so that they will be in the correct order.
     OrderedTombstoneMap<RegionEntry> tombstones = new OrderedTombstoneMap<RegionEntry>();
     if (rm != null) {
       ConcurrentMapWithReusableEntries<Object, Object> other = rm.getCustomEntryConcurrentHashMap();
       Iterator<Map.Entry<Object, Object>> it = other.entrySetWithReusableEntries().iterator();
       while (it.hasNext()) {
         Map.Entry<Object, Object> me = it.next();
         it.remove(); // This removes the RegionEntry from "rm" but it does not decrement its
                      // refcount to an offheap value.
         RegionEntry oldRe = (RegionEntry) me.getValue();
         Object key = me.getKey();

         @Retained
         @Released
         Object value = oldRe
             .getValueRetain((RegionEntryContext) ((AbstractRegionMap) rm)._getOwnerObject(), true);

         try {
           if (value == Token.NOT_AVAILABLE) {
             // fix for bug 43993
             value = null;
           }
           if (value == Token.TOMBSTONE && !_getOwner().getConcurrencyChecksEnabled()) {
             continue;
           }
           RegionEntry newRe =
               getEntryFactory().createEntry((RegionEntryContext) _getOwnerObject(), key, value);
           // TODO: passing value to createEntry causes a problem with the disk stats.
           // The disk stats have already been set to track oldRe.
           // So when we call createEntry we probably want to give it REMOVED_PHASE1
           // and then set the value in copyRecoveredEntry it a way that does not
           // change the disk stats. This also depends on DiskEntry.Helper.initialize not changing
           // the stats for REMOVED_PHASE1
           copyRecoveredEntry(oldRe, newRe);
           // newRe is now in this.getCustomEntryConcurrentHashMap().
           if (newRe.isTombstone()) {
             VersionTag tag = newRe.getVersionStamp().asVersionTag();
             tombstones.put(tag, newRe);
           } else {
             _getOwner().updateSizeOnCreate(key, _getOwner().calculateRegionEntryValueSize(newRe));
             if (_getOwner() instanceof BucketRegionQueue) {
               BucketRegionQueue brq = (BucketRegionQueue) _getOwner();
               brq.incSecondaryQueueSize(1);
             }
           }
           incEntryCount(1);
           lruEntryUpdate(newRe);
         } finally {
           OffHeapHelper.release(value);
           if (oldRe instanceof OffHeapRegionEntry) {
             ((OffHeapRegionEntry) oldRe).release();
           }
         }
         lruUpdateCallback();
       }
     } else {
       for (Iterator<RegionEntry> iter = regionEntries().iterator(); iter.hasNext();) {
         RegionEntry re = iter.next();
         if (re.isTombstone()) {
           if (re.getVersionStamp() == null) { // bug #50992 - recovery from versioned to
                                               // non-versioned
             iter.remove();
             continue;
           } else {
             tombstones.put(re.getVersionStamp().asVersionTag(), re);
           }
         } else {
           _getOwner().updateSizeOnCreate(re.getKey(),
               _getOwner().calculateRegionEntryValueSize(re));
           if (_getOwner() instanceof BucketRegionQueue) {
             BucketRegionQueue brq = (BucketRegionQueue) _getOwner();
             brq.incSecondaryQueueSize(1);
           }
         }
       }
       incEntryCount(size());
       // Since lru was not being done during recovery call it now.
       lruUpdateCallback();
     }

     // Schedule all of the tombstones, now that we have sorted them
     Map.Entry<VersionTag, RegionEntry> entry;
     while ((entry = tombstones.take()) != null) {
       // refresh the tombstone so it doesn't time out too soon
       _getOwner().scheduleTombstone(entry.getValue(), entry.getKey());
     }

   }

   private void copyRecoveredEntry(RegionEntry oldRe, RegionEntry newRe) {
     if (newRe.getVersionStamp() != null) {
       newRe.getVersionStamp().setMemberID(oldRe.getVersionStamp().getMemberID());
       newRe.getVersionStamp().setVersions(oldRe.getVersionStamp().asVersionTag());
     }

     if (newRe instanceof AbstractOplogDiskRegionEntry) {
       ((AbstractOplogDiskRegionEntry) newRe).setDiskId(oldRe);
       _getOwner().getDiskRegion().replaceIncompatibleEntry((DiskEntry) oldRe, (DiskEntry) newRe);
     }
     getEntryMap().put(newRe.getKey(), newRe);
   }

   @Override
   @Retained // Region entry may contain an off-heap value
   public RegionEntry initRecoveredEntry(Object key, DiskEntry.RecoveredEntry value) {
     boolean needsCallback = false;
     @Retained
     RegionEntry newRe =
         getEntryFactory().createEntry((RegionEntryContext) _getOwnerObject(), key, value);
     synchronized (newRe) {
       if (value.getVersionTag() != null && newRe.getVersionStamp() != null) {
         newRe.getVersionStamp().setVersions(value.getVersionTag());
       }
       RegionEntry oldRe = putEntryIfAbsent(key, newRe);
       while (oldRe != null) {
         synchronized (oldRe) {
           if (oldRe.isRemoved() && !oldRe.isTombstone()) {
             if (_isOwnerALocalRegion()) {
               _getOwner().getCachePerfStats().incRetries();
             }
             getEntryMap().remove(key, oldRe);
             oldRe = putEntryIfAbsent(key, newRe);
           }
           /*
            * Entry already exists which should be impossible. Free the current entry (if off-heap)
            * and throw an exception.
            */
           else {
             if (newRe instanceof OffHeapRegionEntry) {
               ((OffHeapRegionEntry) newRe).release();
             }

             throw new IllegalStateException(
                 "Could not recover entry for key " + key + ".  The entry already exists!");
           }
         } // synchronized
       }
       if (_isOwnerALocalRegion()) {
         if (newRe.isTombstone()) {
           // refresh the tombstone so it doesn't time out too soon
           _getOwner().scheduleTombstone(newRe, newRe.getVersionStamp().asVersionTag());
         } else {
           _getOwner().updateSizeOnCreate(key, _getOwner().calculateRegionEntryValueSize(newRe));
         }
         // incEntryCount is called for a tombstone because scheduleTombstone does entryCount--.
         incEntryCount(1); // we are creating an entry that was recovered from disk including
                           // tombstone
       }
       lruEntryUpdate(newRe);
       needsCallback = true;
     }
     if (needsCallback) {
       lruUpdateCallback();
     }

     EntryLogger.logRecovery(_getOwnerObject(), key, value);

     return newRe;
   }

   @Override
   public RegionEntry updateRecoveredEntry(Object key, DiskEntry.RecoveredEntry value) {
     boolean needsCallback = false;
     RegionEntry re = getEntry(key);
     if (re == null) {
       return null;
     }
     synchronized (re) {
       if (re.isRemoved() && !re.isTombstone()) {
         return null;
       }
       if (value.getVersionTag() != null && re.getVersionStamp() != null) {
         re.getVersionStamp().setVersions(value.getVersionTag());
       }
       try {
         if (_isOwnerALocalRegion()) {
           boolean oldValueWasTombstone = re.isTombstone();
           boolean oldIsDestroyedOrRemoved = re.isDestroyedOrRemoved();
           if (oldValueWasTombstone) {
             // when a tombstone is to be overwritten, unschedule it first
             _getOwner().unscheduleTombstone(re);
             // unscheduleTombstone incs entryCount which is ok
             // because we either set the value after this so that
             // the entry exists or we call scheduleTombstone which
             // will dec entryCount.
           }
           final int oldSize = _getOwner().calculateRegionEntryValueSize(re);
           re.setValue(_getOwner(), value); // OFFHEAP no need to call
                                            // AbstractRegionMap.prepareValueForCache because
                                            // setValue is overridden for disk and that code takes
                                            // apart value (RecoveredEntry) and prepares its nested
                                            // value for the cache
           if (re.isTombstone()) {
             _getOwner().scheduleTombstone(re, re.getVersionStamp().asVersionTag());
             if (!oldIsDestroyedOrRemoved) {
               _getOwner().updateSizeOnRemove(key, oldSize);
             }
           } else if (oldIsDestroyedOrRemoved) {
             _getOwner().updateSizeOnCreate(key, _getOwner().calculateRegionEntryValueSize(re));
           } else {
             _getOwner().updateSizeOnPut(key, oldSize,
                 _getOwner().calculateRegionEntryValueSize(re));
           }
         } else {
           value.applyToDiskEntry((PlaceHolderDiskRegion) _getOwnerObject(), (DiskEntry) re,
               (RegionEntryContext) _getOwnerObject());
         }
       } catch (RegionClearedException rce) {
         throw new IllegalStateException(
             "RegionClearedException should never happen in this context", rce);
       }
       lruEntryUpdate(re);
       needsCallback = true;
     }
     if (needsCallback) {
       lruUpdateCallback();
     }

     EntryLogger.logRecovery(_getOwnerObject(), key, value);

     return re;
   }

   @Override
   public boolean initialImagePut(final Object key, final long lastModified, Object newValue,
       final boolean wasRecovered, boolean deferLRUCallback,
       VersionTag entryVersion,
       InternalDistributedMember sender, boolean isSynchronizing) {
     boolean result = false;
     boolean done = false;
     boolean cleared = false;
     final LocalRegion owner = _getOwner();
     boolean acceptedVersionTag = entryVersion != null && owner.getConcurrencyChecksEnabled();

     if (newValue == Token.TOMBSTONE && !owner.getConcurrencyChecksEnabled()) {
       return false;
     }

     if (owner instanceof HARegion && newValue instanceof CachedDeserializable) {
       newValue = ((HARegion) owner).updateHAEventWrapper(sender, (CachedDeserializable) newValue);
       if (newValue == null) {
         return false;
       }
     }

     try {
       RegionEntry newRe = getEntryFactory().createEntry(owner, key, Token.REMOVED_PHASE1);
       RegionEntry oldRe = null;
       synchronized (newRe) {
         try {
           oldRe = putEntryIfAbsent(key, newRe);
           while (!done && oldRe != null) {
             synchronized (oldRe) {
               if (oldRe.isRemovedPhase2()) {
                 owner.getCachePerfStats().incRetries();
                 getEntryMap().remove(key, oldRe);
                 oldRe = putEntryIfAbsent(key, newRe);
               } else {
                 if (acceptedVersionTag) {
                   Assert.assertTrue(entryVersion.getMemberID() != null,
                       "GII entry versions must have identifiers");
                   boolean isTombstone = (newValue == Token.TOMBSTONE);
                   // don't reschedule the tombstone if it hasn't changed
                   boolean isSameTombstone = oldRe.isTombstone() && isTombstone
                       && oldRe.getVersionStamp().asVersionTag().equals(entryVersion);
                   if (isSameTombstone) {
                     return true;
                   }
                   processVersionTagForGII(oldRe, owner, entryVersion, isTombstone, sender,
                       !wasRecovered || isSynchronizing);

                 }
                 final boolean oldIsTombstone = oldRe.isTombstone();
                 final boolean oldIsDestroyedOrRemoved = oldRe.isDestroyedOrRemoved();
                 final int oldSize = owner.calculateRegionEntryValueSize(oldRe);
                 if (owner.getIndexManager() != null) {
                   // Due to having no reverse map, we need to be able to generate the oldkey
                   // before doing an update
                   // Without the BEFORE_UPDATE_OP, we would see duplicate entries in the index
                   // as the update could not locate the old key
                   if (!oldRe.isRemoved()) {
                     owner.getIndexManager().updateIndexes(oldRe, IndexManager.REMOVE_ENTRY,
                         IndexProtocol.BEFORE_UPDATE_OP);
                   }
                 }
                 result = oldRe.initialImagePut(owner, lastModified, newValue, wasRecovered,
                     acceptedVersionTag);
                 if (result) {
                   if (oldIsTombstone) {
                     owner.unscheduleTombstone(oldRe);
                     if (newValue != Token.TOMBSTONE) {
                       lruEntryCreate(oldRe);
                     } else {
                       lruEntryUpdate(oldRe);
                     }
                   }
                   if (newValue == Token.TOMBSTONE) {
                     if (!oldIsDestroyedOrRemoved) {
                       owner.updateSizeOnRemove(key, oldSize);
                     }
                     if (owner.getServerProxy() == null
                         && owner.getVersionVector().isTombstoneTooOld(
                             entryVersion.getMemberID(), entryVersion.getRegionVersion())) {
                       // the received tombstone has already been reaped, so don't retain it
                       if (owner.getIndexManager() != null) {
                         owner.getIndexManager().updateIndexes(oldRe, IndexManager.REMOVE_ENTRY,
                             IndexProtocol.REMOVE_DUE_TO_GII_TOMBSTONE_CLEANUP);
                       }
                       removeTombstone(oldRe, entryVersion, false, false);
                       return false;
                     } else {
                       owner.scheduleTombstone(oldRe, entryVersion);
                       lruEntryDestroy(oldRe);
                     }
                   } else {
                     int newSize = owner.calculateRegionEntryValueSize(oldRe);
                     if (!oldIsTombstone) {
                       owner.updateSizeOnPut(key, oldSize, newSize);
                     } else {
                       owner.updateSizeOnCreate(key, newSize);
                     }
                     EntryLogger.logInitialImagePut(_getOwnerObject(), key, newValue);
                   }
                 }
                 if (owner.getIndexManager() != null) {
                   // if existing/current re is a tombstone - note oldRe at this point is currentRe
                   if (oldRe.isRemoved()) {
                     owner.getIndexManager().updateIndexes(oldRe, IndexManager.REMOVE_ENTRY,
                         IndexProtocol.REMOVE_DUE_TO_GII_TOMBSTONE_CLEANUP);
                   } else {
                     owner.getIndexManager().updateIndexes(oldRe,
                         oldIsDestroyedOrRemoved ? IndexManager.ADD_ENTRY
                             : IndexManager.UPDATE_ENTRY,
                         oldIsDestroyedOrRemoved ? IndexProtocol.OTHER_OP
                             : IndexProtocol.AFTER_UPDATE_OP);
                   }
                 }
                 done = true;

               }
             }
           }
           if (!done) {
             if (acceptedVersionTag) {
               Assert.assertTrue(entryVersion.getMemberID() != null,
                   "GII entry versions must have identifiers");
               boolean isTombstone = (newValue == Token.TOMBSTONE);
               processVersionTagForGII(newRe, owner, entryVersion, isTombstone, sender,
                   !wasRecovered || isSynchronizing);
             }
             result = newRe.initialImageInit(owner, lastModified, newValue, true, wasRecovered,
                 acceptedVersionTag);
             if (result) {
               if (newValue == Token.TOMBSTONE) {
                 owner.scheduleTombstone(newRe, entryVersion);
               } else {
                 owner.updateSizeOnCreate(key, owner.calculateRegionEntryValueSize(newRe));
                 EntryLogger.logInitialImagePut(_getOwnerObject(), key, newValue);
                 lruEntryCreate(newRe);
               }
               incEntryCount(1);
             }

             // Update local indexes
             if (owner.getIndexManager() != null && !newRe.isRemoved()) {
               owner.getIndexManager().updateIndexes(newRe,
                   IndexManager.ADD_ENTRY,
                   IndexProtocol.OTHER_OP);
             }
             done = true;

           }
         } catch (ConcurrentCacheModificationException e) {
           // We do not want to do any clean up of indexes because it is assumed that
           // the cause of the concurrent modification would have updated the indexes appropriately
           return false;
         } finally {
           if (done && result) {
             if (owner instanceof BucketRegionQueue) {
               BucketRegionQueue brq = (BucketRegionQueue) owner;
               brq.addToEventQueue(key, done, null);
             }
           }
           if (!done) {
             removeEntry(key, newRe, false);
             // Update local indexes
             if (owner.getIndexManager() != null && !newRe.isRemoved()) {
               // attempt to clean up any thread local state,
               // not intended to actually do any removal
               try {
                 owner.getIndexManager().updateIndexes(newRe,
                     IndexManager.REMOVE_ENTRY,
                     IndexProtocol.CLEAN_UP_THREAD_LOCALS);
               } catch (QueryException qe) {
                 logger.info("Unable to clean up thread locals for indexes", qe);
               }
             }
           }
         }
       } // synchronized
     } catch (RegionClearedException rce) {
       done = false;
       cleared = true;
     } catch (QueryException qe) {
       done = false;
       cleared = true;
     } finally {
       if (done && !deferLRUCallback) {
         lruUpdateCallback();
       } else if (!cleared) {
         resetThreadLocals();
       }
     }

     return result;
   }

   @Override
   public boolean confirmEvictionDestroy(RegionEntry regionEntry) {
     /* We arn't in an LRU context, and should never get here */
     Assert.assertTrue(false, "Not an LRU region, can not confirm LRU eviction operation");
     return true;
   }

   @Override
   public boolean destroy(EntryEventImpl event, boolean inTokenMode, boolean duringRI,
       boolean cacheWrite, boolean isEviction, Object expectedOldValue, boolean removeRecoveredEntry)
       throws CacheWriterException, EntryNotFoundException, TimeoutException {
     RegionMapDestroy regionMapDestroy = new RegionMapDestroy((InternalRegion) owner, this, this);
     return regionMapDestroy.destroy(event, inTokenMode, duringRI, cacheWrite, isEviction,
         expectedOldValue, removeRecoveredEntry);
   }

   @Override
   public void txApplyDestroy(Object key, TransactionId txId, TXRmtEvent txEvent,
       boolean inTokenMode, boolean inRI, Operation op, EventID eventId, Object aCallbackArgument,
       List<EntryEventImpl> pendingCallbacks, FilterRoutingInfo filterRoutingInfo,
       ClientProxyMembershipID bridgeContext, boolean isOriginRemote, TXEntryState txEntryState,
       VersionTag versionTag, long tailKey) {
     assert pendingCallbacks != null;
     final boolean isDebugEnabled = logger.isDebugEnabled();

     final LocalRegion owner = _getOwner();

     final boolean isRegionReady = !inTokenMode;
     final boolean hasRemoteOrigin = !txId.getMemberId().equals(owner.getMyId());
     boolean callbackEventAddedToPending = false;
     IndexManager oqlIndexManager = owner.getIndexManager();
     final boolean locked = owner.lockWhenRegionIsInitializing();
     try {
       RegionEntry re = getEntry(key);
       if (re != null) {
         // Fix for Bug #44431. We do NOT want to update the region and wait
         // later for index INIT as region.clear() can cause inconsistency if
         // happened in parallel as it also does index INIT.
         if (oqlIndexManager != null) {
           oqlIndexManager.waitForIndexInit();
         }
         try {
           synchronized (re) {
             if (!re.isRemoved() || re.isTombstone()) {
               Object oldValue = re.getValueInVM(owner);
               final int oldSize = owner.calculateRegionEntryValueSize(re);
               final boolean wasDestroyedOrRemoved = re.isDestroyedOrRemoved();
               // Create an entry event only if the calling context is
               // a receipt of a TXCommitMessage AND there are callbacks installed
               // for this region
               @Released
               final EntryEventImpl callbackEvent = txCallbackEventFactory
                   .createCallbackEvent(owner, op, key, null, txId,
                       txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                       txEntryState, versionTag, tailKey);
               try {

                 if (owner.isUsedForPartitionedRegionBucket()) {
                   txHandleWANEvent(owner, callbackEvent, txEntryState);
                 }
                 callbackEvent.setRegionEntry(re);
                 callbackEvent.setOldValue(oldValue);
                 if (isDebugEnabled) {
                   logger.debug("txApplyDestroy callbackEvent={}", callbackEvent);
                 }

                 txRemoveOldIndexEntry(Operation.DESTROY, re);
                 if (txEvent != null) {
                   txEvent.addDestroy(owner, re, re.getKey(), aCallbackArgument);
                 }
                 boolean clearOccured = false;
                 try {
                   processAndGenerateTXVersionTag(callbackEvent, re, txEntryState);
                   if (inTokenMode) {
                     if (oldValue == Token.TOMBSTONE) {
                       owner.unscheduleTombstone(re);
                     }
                     re.setValue(owner, Token.DESTROYED);
                   } else {
                     if (!re.isTombstone()) {
                       {
                         if (owner.getConcurrencyChecksEnabled()
                             && callbackEvent.getVersionTag() != null) {
                           re.makeTombstone(owner, callbackEvent.getVersionTag());
                         } else {
                           re.removePhase1(owner, false); // fix for bug 43063
                           re.removePhase2();
                           removeEntry(key, re, false);
                         }
                       }
                     } else {
                       owner.rescheduleTombstone(re, re.getVersionStamp().asVersionTag());
                     }
                   }
                   EntryLogger.logTXDestroy(_getOwnerObject(), key);
                   if (!wasDestroyedOrRemoved) {
                     owner.updateSizeOnRemove(key, oldSize);
                   }
                 } catch (RegionClearedException rce) {
                   clearOccured = true;
                 }
                 owner.txApplyDestroyPart2(re, re.getKey(), inTokenMode,
                     clearOccured /* Clear Conflciting with the operation */,
                     wasDestroyedOrRemoved);
                 boolean invokeCallbacks = shouldInvokeCallbacks(owner, isRegionReady || inRI);
                 if (invokeCallbacks) {
                   switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                   pendingCallbacks.add(callbackEvent);
                   callbackEventAddedToPending = true;
                 }
                 if (!clearOccured) {
                   lruEntryDestroy(re);
                 }
                 if (owner.getConcurrencyChecksEnabled() && txEntryState != null) {
                   txEntryState.setVersionTag(callbackEvent.getVersionTag());
                 }
               } finally {
                 if (!callbackEventAddedToPending)
                   releaseEvent(callbackEvent);
               }
             }
           }
         } finally {
           if (oqlIndexManager != null) {
             oqlIndexManager.countDownIndexUpdaters();
           }
         }
       } else if (inTokenMode || owner.getConcurrencyChecksEnabled()) {
         // treating tokenMode and re == null as same, since we now want to
         // generate versions and Tombstones for destroys
         boolean dispatchListenerEvent = inTokenMode;
         boolean opCompleted = false;
         RegionEntry newRe = getEntryFactory().createEntry(owner, key, Token.REMOVED_PHASE1);
         if (oqlIndexManager != null) {
           oqlIndexManager.waitForIndexInit();
         }
         EntryEventImpl callbackEvent = null;
         try {
           synchronized (newRe) {
             RegionEntry oldRe = putEntryIfAbsent(key, newRe);
             while (!opCompleted && oldRe != null) {
               synchronized (oldRe) {
                 if (oldRe.isRemovedPhase2()) {
                   owner.getCachePerfStats().incRetries();
                   getEntryMap().remove(key, oldRe);
                   oldRe = putEntryIfAbsent(key, newRe);
                 } else {
                   try {
                     boolean invokeCallbacks = shouldInvokeCallbacks(owner, isRegionReady || inRI);
                     callbackEvent = txCallbackEventFactory
                         .createCallbackEvent(owner, op, key, null, txId, txEvent,
                             eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                             txEntryState,
                             versionTag, tailKey);
                     try {
                       callbackEvent.setRegionEntry(oldRe);
                       callbackEvent.setOldValue(Token.NOT_AVAILABLE);
                       if (isDebugEnabled) {
                         logger.debug("txApplyDestroy token mode callbackEvent={}", callbackEvent);
                       }
                       if (owner.isUsedForPartitionedRegionBucket()) {
                         txHandleWANEvent(owner, callbackEvent, txEntryState);
                       }
                       processAndGenerateTXVersionTag(callbackEvent, oldRe, txEntryState);
                       if (invokeCallbacks) {
                         switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                         pendingCallbacks.add(callbackEvent);
                         callbackEventAddedToPending = true;
                       }
                       int oldSize = 0;
                       boolean wasTombstone = oldRe.isTombstone();
                       boolean wasDestroyedOrRemoved = oldRe.isDestroyedOrRemoved();
                       {
                         if (!wasTombstone) {
                           oldSize = owner.calculateRegionEntryValueSize(oldRe);
                         }
                       }
                       // TODO: Token.DESTROYED should only be used if "inTokenMode".
                       // Otherwise this should be a TOMBSTONE
                       oldRe.setValue(owner, Token.DESTROYED);
                       EntryLogger.logTXDestroy(_getOwnerObject(), key);
                       if (wasTombstone) {
                         owner.unscheduleTombstone(oldRe);
                       }
                       if (!wasDestroyedOrRemoved) {
                         owner.updateSizeOnRemove(oldRe.getKey(), oldSize);
                       }
                       owner.txApplyDestroyPart2(oldRe, oldRe.getKey(), inTokenMode,
                           false /* Clear Conflicting with the operation */, wasDestroyedOrRemoved);
                       lruEntryDestroy(oldRe);
                     } finally {
                       if (!callbackEventAddedToPending)
                         releaseEvent(callbackEvent);
                     }
                   } catch (RegionClearedException rce) {
                     owner.txApplyDestroyPart2(oldRe, oldRe.getKey(), inTokenMode,
                         true /* Clear Conflicting with the operation */, true);
                   }
                   if (owner.getConcurrencyChecksEnabled()
                       && callbackEvent.getVersionTag() != null) {
                     oldRe.makeTombstone(owner, callbackEvent.getVersionTag());
                   } else if (!inTokenMode) {
                     // only remove for NORMAL regions if they do not generate versions see 51781
                     oldRe.removePhase1(owner, false); // fix for bug 43063
                     oldRe.removePhase2();
                     removeEntry(key, oldRe, false);
                   }
                   opCompleted = true;
                 }
               }
             }
             if (!opCompleted) {
               opCompleted = true;
               boolean invokeCallbacks = shouldInvokeCallbacks(owner, isRegionReady || inRI);
               callbackEvent = txCallbackEventFactory
                   .createCallbackEvent(owner, op, key, null, txId, txEvent, eventId,
                       aCallbackArgument, filterRoutingInfo, bridgeContext, txEntryState, versionTag,
                       tailKey);
               try {
                 callbackEvent.setRegionEntry(newRe);
                 callbackEvent.setOldValue(Token.NOT_AVAILABLE);
                 if (isDebugEnabled) {
                   logger.debug("txApplyDestroy token mode callbackEvent={}", callbackEvent);
                 }
                 if (owner.isUsedForPartitionedRegionBucket()) {
                   txHandleWANEvent(owner, callbackEvent, txEntryState);
                 }
                 processAndGenerateTXVersionTag(callbackEvent, newRe, txEntryState);
                 if (invokeCallbacks) {
                   switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                   pendingCallbacks.add(callbackEvent);
                   callbackEventAddedToPending = true;
                 }
                 EntryLogger.logTXDestroy(_getOwnerObject(), key);
                 if (owner.getConcurrencyChecksEnabled() && callbackEvent.getVersionTag() != null) {
                   newRe.makeTombstone(owner, callbackEvent.getVersionTag());
                 } else if (!inTokenMode) {
                   // only remove for NORMAL regions if they do not generate versions see 51781
                   newRe.removePhase1(owner, false); // fix for bug 43063
                   newRe.removePhase2();
                   removeEntry(key, newRe, false);
                 } else {
                   newRe.setValue(owner, Token.DESTROYED);
                 }
                 owner.txApplyDestroyPart2(newRe, newRe.getKey(), inTokenMode,
                     false /* clearConflict */, true);
                 // Note no need for LRU work since the entry is destroyed
                 // and will be removed when gii completes
               } finally {
                 if (!callbackEventAddedToPending)
                   releaseEvent(callbackEvent);
               }
             }
             if (owner.getConcurrencyChecksEnabled() && txEntryState != null) {
               txEntryState.setVersionTag(callbackEvent.getVersionTag());
             }
           }
         } catch (RegionClearedException e) {
           // TODO
         } finally {
           if (oqlIndexManager != null) {
             oqlIndexManager.countDownIndexUpdaters();
           }
         }
       } else { // re == null
         // Fix bug#43594
         // In cases where bucket region is re-created, it may so happen that
         // the destroy is already applied on the Initial image provider, thus
         // causing region entry to be absent.
         // Notify clients with client events.
         @Released
         EntryEventImpl callbackEvent =
             txCallbackEventFactory
                 .createCallbackEvent(owner, op, key, null, txId, txEvent, eventId,
                     aCallbackArgument,
                     filterRoutingInfo, bridgeContext, txEntryState, versionTag, tailKey);
         try {
           if (owner.isUsedForPartitionedRegionBucket()) {
             txHandleWANEvent(owner, callbackEvent, txEntryState);
           }
           switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
           pendingCallbacks.add(callbackEvent);
           callbackEventAddedToPending = true;
         } finally {
           if (!callbackEventAddedToPending)
             releaseEvent(callbackEvent);
         }
       }
     } catch (DiskAccessException dae) {
       owner.handleDiskAccessException(dae);
       throw dae;
     } finally {
       if (locked) {
         owner.unlockWhenRegionIsInitializing();
       }
     }
   }

   void releaseEvent(final EntryEventImpl event) {
     event.release();
   }

   @Override
   public boolean invalidate(EntryEventImpl event, boolean invokeCallbacks, boolean forceNewEntry,
       boolean forceCallbacks) throws EntryNotFoundException {
     final boolean isDebugEnabled = logger.isDebugEnabled();

     final LocalRegion owner = _getOwner();
     if (owner == null) {
       // "fix" for bug 32440
       Assert.assertTrue(false, "The owner for RegionMap " + this + " is null for event " + event);

     }
     logger.debug("ARM.invalidate invoked for key {}", event.getKey());
     boolean didInvalidate = false;
     RegionEntry invalidatedRe = null;
     boolean clearOccured = false;
     DiskRegion dr = owner.getDiskRegion();
     boolean ownerIsInitialized = owner.isInitialized();

     // Fix for Bug #44431. We do NOT want to update the region and wait
     // later for index INIT as region.clear() can cause inconsistency if
     // happened in parallel as it also does index INIT.
     IndexManager oqlIndexManager = owner.getIndexManager();
     if (oqlIndexManager != null) {
       oqlIndexManager.waitForIndexInit();
     }
     lockForCacheModification(owner, event);
     final boolean locked = owner.lockWhenRegionIsInitializing();
     try {
       try {
         try {
           if (forceNewEntry || forceCallbacks) {
             boolean opCompleted = false;
             RegionEntry newRe =
                 getEntryFactory().createEntry(owner, event.getKey(), Token.REMOVED_PHASE1);
             synchronized (newRe) {
               try {
                 RegionEntry oldRe = putEntryIfAbsent(event.getKey(), newRe);

                 while (!opCompleted && oldRe != null) {
                   synchronized (oldRe) {
                     // if the RE is in phase 2 of removal, it will really be removed
                     // from the map. Otherwise, we can use it here and the thread
                     // that is destroying the RE will see the invalidation and not
                     // proceed to phase 2 of removal.
                     if (oldRe.isRemovedPhase2()) {
                       owner.getCachePerfStats().incRetries();
                       getEntryMap().remove(event.getKey(), oldRe);
                       oldRe = putEntryIfAbsent(event.getKey(), newRe);
                     } else {
                       opCompleted = true;
                       event.setRegionEntry(oldRe);
                       if (oldRe.isDestroyed()) {
                         if (isDebugEnabled) {
                           logger.debug(
                               "mapInvalidate: Found DESTROYED token, not invalidated; key={}",
                               event.getKey());
                         }
                       } else if (oldRe.isInvalid()) {

                         // was already invalid, do not invoke listeners or increment stat
                         handleAlreadyInvalidEntry(event, owner, oldRe);
                         try {
                           oldRe.setValue(owner, oldRe.getValueInVM(owner)); // OFFHEAP noop setting
                                                                             // an already invalid to
                                                                             // invalid; No need to
                                                                             // call
                                                                             // prepareValueForCache
                                                                             // since it is an
                                                                             // invalid token.
                         } catch (RegionClearedException e) {
                           // that's okay - when writing an invalid into a disk, the
                           // region has been cleared (including this token)
                         }
                       } else {
                         owner.serverInvalidate(event);
                         if (owner.getConcurrencyChecksEnabled()
                             && event.noVersionReceivedFromServer()) {
                           // server did not perform the invalidation, so don't leave an invalid
                           // entry here
                           return false;
                         }
                         final int oldSize = owner.calculateRegionEntryValueSize(oldRe);
                         // added for cq which needs old value. rdubey
                         FilterProfile fp = owner.getFilterProfile();
                         if (!oldRe.isRemoved() && (fp != null && fp.getCqCount() > 0)) {

                           Object oldValue = oldRe.getValueInVM(owner); // OFFHEAP EntryEventImpl
                                                                        // oldValue

                           // this will not fault in the value.
                           if (oldValue == Token.NOT_AVAILABLE) {
                             event.setOldValue(oldRe.getValueOnDiskOrBuffer(owner));
                           } else {
                             event.setOldValue(oldValue);
                           }
                         }
                         boolean isCreate = false;
                         try {
                           if (oldRe.isRemoved()) {
                             processVersionTag(oldRe, event);
                             event.putNewEntry(owner, oldRe);
                             EntryLogger.logInvalidate(event);
                             owner.recordEvent(event);
                             if (!oldRe.isTombstone()) {
                               owner.updateSizeOnPut(event.getKey(), oldSize,
                                   event.getNewValueBucketSize());
                             } else {
                               owner.updateSizeOnCreate(event.getKey(),
                                   event.getNewValueBucketSize());
                               isCreate = true;
                             }
                           } else {
                             processVersionTag(oldRe, event);
                             event.putExistingEntry(owner, oldRe);
                             EntryLogger.logInvalidate(event);
                             owner.recordEvent(event);
                             owner.updateSizeOnPut(event.getKey(), oldSize,
                                 event.getNewValueBucketSize());
                           }
                         } catch (RegionClearedException e) {
                           // generate versionTag for the event
                           EntryLogger.logInvalidate(event);
                           owner.recordEvent(event);
                           clearOccured = true;
                         }
                         owner.basicInvalidatePart2(oldRe, event,
                             clearOccured /* conflict with clear */, invokeCallbacks);
                         if (!clearOccured) {
                           if (isCreate) {
                             lruEntryCreate(oldRe);
                           } else {
                             lruEntryUpdate(oldRe);
                           }
                         }
                         didInvalidate = true;
                         invalidatedRe = oldRe;
                       }
                     }
                   } // synchronized oldRe
                 } // while oldRe exists

                 if (!opCompleted) {
                   if (forceNewEntry && event.isFromServer()) {
                     // don't invoke listeners - we didn't force new entries for
                     // CCU invalidations before 7.0, and listeners don't care
                     if (!FORCE_INVALIDATE_EVENT) {
                       event.inhibitCacheListenerNotification(true);
                     }
                   }
                   event.setRegionEntry(newRe);
                   owner.serverInvalidate(event);
                   if (!forceNewEntry && event.noVersionReceivedFromServer()) {
                     // server did not perform the invalidation, so don't leave an invalid
                     // entry here
                     return false;
                   }
                   try {
                     ownerIsInitialized = owner.isInitialized();
                     if (!ownerIsInitialized && owner.getDataPolicy().withReplication()) {
                       final int oldSize = owner.calculateRegionEntryValueSize(newRe);
                       invalidateEntry(event, newRe, oldSize);
                     } else {
                       invalidateNewEntry(event, owner, newRe);
                     }
                   } catch (RegionClearedException e) {
                     // TODO: deltaGII: do we even need RegionClearedException?
                     // generate versionTag for the event
                     owner.recordEvent(event);
                     clearOccured = true;
                   }
                   owner.basicInvalidatePart2(newRe, event, clearOccured /* conflict with clear */,
                       invokeCallbacks);
                   if (!clearOccured) {
                     lruEntryCreate(newRe);
                     incEntryCount(1);
                   }
                   opCompleted = true;
                   didInvalidate = true;
                   invalidatedRe = newRe;
                   // Don't leave an entry in the cache, if we
                   // just wanted to force the distribution and events
                   // for this invalidate
                   if (!forceNewEntry) {
                     removeEntry(event.getKey(), newRe, false);
                   }
                 } // !opCompleted
               } catch (ConcurrentCacheModificationException ccme) {
                 VersionTag tag = event.getVersionTag();
                 if (tag != null && tag.isTimeStampUpdated()) {
                   // Notify gateways of new time-stamp.
                   owner.notifyTimestampsToGateways(event);
                 }
                 throw ccme;
               } finally {
                 if (!opCompleted) {
                   removeEntry(event.getKey(), newRe, false);
                 }
               }
             } // synchronized newRe
           } // forceNewEntry
           else { // !forceNewEntry
             boolean retry = true;
             // RegionEntry retryEntry = null;
             // int retries = -1;

             while (retry) {
               retry = false;
               boolean entryExisted = false;
               RegionEntry re = getEntry(event.getKey());
               RegionEntry tombstone = null;
               boolean haveTombstone = false;
               if (re != null && re.isTombstone()) {
                 tombstone = re;
                 haveTombstone = true;
                 re = null;
               }
               if (re == null) {
                 ownerIsInitialized = owner.isInitialized();
                 if (!ownerIsInitialized) {
                   // when GII message arrived or processed later than invalidate
                   // message, the entry should be created as placeholder
                   RegionEntry newRe = haveTombstone ? tombstone
                       : getEntryFactory().createEntry(owner, event.getKey(), Token.INVALID);
                   synchronized (newRe) {
                     if (haveTombstone && !tombstone.isTombstone()) {
                       // state of the tombstone has changed so we need to retry
                       retry = true;
                       // retryEntry = tombstone; // leave this in place for debugging
                       continue;
                     }
                     re = putEntryIfAbsent(event.getKey(), newRe);
                     if (re == tombstone) {
                       re = null; // pretend we don't have an entry
                     }
                   }
                 } else if (owner.getServerProxy() != null) {
                   Object sync = haveTombstone ? tombstone : new Object();
                   synchronized (sync) {
                     if (haveTombstone && !tombstone.isTombstone()) {
                       // bug 45295: state of the tombstone has changed so we need to retry
                       retry = true;
                       // retryEntry = tombstone; // leave this in place for debugging
                       continue;
                     }

                     // bug #43287 - send event to server even if it's not in the client (LRU may
                     // have evicted it)
                     owner.serverInvalidate(event);
                     if (owner.getConcurrencyChecksEnabled()) {
                       if (event.getVersionTag() == null) {
                         // server did not perform the invalidation, so don't leave an invalid
                         // entry here
                         return false;
                       } else if (tombstone != null) {
                         processVersionTag(tombstone, event);
                         try {
                           tombstone.setValue(owner, Token.TOMBSTONE);
                         } catch (RegionClearedException e) {
                           // that's okay - when writing a tombstone into a disk, the
                           // region has been cleared (including this tombstone)
                         } catch (ConcurrentCacheModificationException ccme) {
                           VersionTag tag = event.getVersionTag();
                           if (tag != null && tag.isTimeStampUpdated()) {
                             // Notify gateways of new time-stamp.
                             owner.notifyTimestampsToGateways(event);
                           }
                           throw ccme;
                         }
                         // update the tombstone's version to prevent an older CCU/putAll from
                         // overwriting it
                         owner.rescheduleTombstone(tombstone, event.getVersionTag());
                       }
                     }
                   }
                   entryExisted = true;
                 }
               }
               if (re != null) {
                 // Gester: Race condition in GII
                 // when adding the placeholder for invalidate entry during GII,
                 // if the GII got processed earlier for this entry, then do
                 // normal invalidate operation
                 synchronized (re) {
                   if (!event.isOriginRemote() && event.getOperation().isExpiration()) {
                     // If this expiration started locally then only do it if the RE is not being
                     // used by a tx.
                     if (re.isInUseByTransaction()) {
                       return false;
                     }
                   }
                   if (re.isTombstone() || (!re.isRemoved() && !re.isDestroyed())) {
                     entryExisted = true;
                     if (re.isInvalid()) {
                       // was already invalid, do not invoke listeners or increment
                       // stat
                       handleAlreadyInvalidEntry(event, owner, re);
                     } else { // previous value not invalid
                       event.setRegionEntry(re);
                       owner.serverInvalidate(event);
                       if (owner.getConcurrencyChecksEnabled()
                           && event.noVersionReceivedFromServer()) {
                         // server did not perform the invalidation, so don't leave an invalid
                         // entry here
                         if (isDebugEnabled) {
                           logger.debug(
                               "returning early because server did not generate a version stamp for this event:{}",
                               event);
                         }
                         return false;
                       }
                       // in case of overflow to disk we need the old value for cqs.
                       if (owner.getFilterProfile().getCqCount() > 0) {
                         // use to be getValue and can cause dead lock rdubey.
                         if (re.isValueNull()) {
                           event.setOldValue(re.getValueOnDiskOrBuffer(owner));
                         } else {
                           Object v = re.getValueInVM(owner);
                           event.setOldValue(v); // OFFHEAP escapes to EntryEventImpl oldValue
                         }
                       }
                       final boolean oldWasTombstone = re.isTombstone();
                       final int oldSize = _getOwner().calculateRegionEntryValueSize(re);
                       try {
                         invalidateEntry(event, re, oldSize);
                       } catch (RegionClearedException rce) {
                         // generate versionTag for the event
                         EntryLogger.logInvalidate(event);
                         _getOwner().recordEvent(event);
                         clearOccured = true;
                       } catch (ConcurrentCacheModificationException ccme) {
                         VersionTag tag = event.getVersionTag();
                         if (tag != null && tag.isTimeStampUpdated()) {
                           // Notify gateways of new time-stamp.
                           owner.notifyTimestampsToGateways(event);
                         }
                         throw ccme;
                       }
                       owner.basicInvalidatePart2(re, event, clearOccured /* conflict with clear */,
                           invokeCallbacks);
                       if (!clearOccured) {
                         if (oldWasTombstone) {
                           lruEntryCreate(re);
                         } else {
                           lruEntryUpdate(re);
                         }
                       }
                       didInvalidate = true;
                       invalidatedRe = re;
                     } // previous value not invalid
                   }
                 } // synchronized re
               } // re != null
               else {
                 // At this point, either it's not in GII mode, or the placeholder
                 // is in region, do nothing
               }
               if (!entryExisted) {
                 owner.checkEntryNotFound(event.getKey());
               }
             } // while(retry)
           } // !forceNewEntry
         } catch (DiskAccessException dae) {
           invalidatedRe = null;
           didInvalidate = false;
           this._getOwner().handleDiskAccessException(dae);
           throw dae;
         } finally {
           if (oqlIndexManager != null) {
             oqlIndexManager.countDownIndexUpdaters();
           }
           if (invalidatedRe != null) {
             owner.basicInvalidatePart3(invalidatedRe, event, invokeCallbacks);
           }
           if (didInvalidate && !clearOccured) {
             try {
               lruUpdateCallback();
             } catch (DiskAccessException dae) {
               this._getOwner().handleDiskAccessException(dae);
               throw dae;
             }
           } else if (!didInvalidate) {
             resetThreadLocals();
           }
         }
         return didInvalidate;
       } finally {
         if (ownerIsInitialized) {
           forceInvalidateEvent(event, owner);
         }
       }
     } finally {
       if (locked) {
         owner.unlockWhenRegionIsInitializing();
       }
       releaseCacheModificationLock(owner, event);
     }

   }

   /**
    * If an entry is already invalid we still want to perform a conflict check, update
    * the entry's version stamp and invoke listeners.
    */
   private void handleAlreadyInvalidEntry(EntryEventImpl event, LocalRegion owner, RegionEntry re) {
     if (logger.isDebugEnabled()) {
       logger.debug("Invalidate: Entry already invalid: '{}'", event.getKey());
     }
     processVersionTag(re, event);
     if (owner.getConcurrencyChecksEnabled() && event.hasValidVersionTag()) {
       // notify clients so they can update their version stamps
       event.invokeCallbacks(owner, true, true);
     }
   }

   private void invalidateNewEntry(EntryEventImpl event, final LocalRegion owner, RegionEntry newRe)
       throws RegionClearedException {
     processVersionTag(newRe, event);
     event.putNewEntry(owner, newRe);
     owner.recordEvent(event);
     owner.updateSizeOnCreate(event.getKey(), event.getNewValueBucketSize());
   }

   protected void invalidateEntry(EntryEventImpl event, RegionEntry re, int oldSize)
       throws RegionClearedException {
     processVersionTag(re, event);
     event.putExistingEntry(_getOwner(), re);
     EntryLogger.logInvalidate(event);
     _getOwner().recordEvent(event);
     _getOwner().updateSizeOnPut(event.getKey(), oldSize, event.getNewValueBucketSize());
   }


   /*
    * (non-Javadoc)
    *
    * @see
    * org.apache.geode.internal.cache.RegionMap#updateEntryVersion(org.apache.geode.internal.cache.
    * EntryEventImpl)
    */
   @Override
   public void updateEntryVersion(EntryEventImpl event) throws EntryNotFoundException {

     final LocalRegion owner = _getOwner();
     if (owner == null) {
       // "fix" for bug 32440
       Assert.assertTrue(false, "The owner for RegionMap " + this + " is null for event " + event);

     }

     DiskRegion dr = owner.getDiskRegion();
     if (dr != null) {
       dr.setClearCountReference();
     }

     lockForCacheModification(owner, event);
     final boolean locked = owner.lockWhenRegionIsInitializing();

     try {
       RegionEntry re = getEntry(event.getKey());

       boolean entryExisted = false;

       if (re != null) {
         // process version tag
         synchronized (re) {

           try {
             if (re.isTombstone() || (!re.isRemoved() && !re.isDestroyed())) {
               entryExisted = true;
             }
             processVersionTag(re, event);
             owner.generateAndSetVersionTag(event, re);
             EntryLogger.logUpdateEntryVersion(event);
             _getOwner().recordEvent(event);
           } catch (ConcurrentCacheModificationException ccme) {
             // Do nothing.
           }

         }
       }
       if (!entryExisted) {
         owner.checkEntryNotFound(event.getKey());
       }
     } catch (DiskAccessException dae) {
       this._getOwner().handleDiskAccessException(dae);
       throw dae;
     } finally {
       if (locked) {
         owner.unlockWhenRegionIsInitializing();
       }
       releaseCacheModificationLock(owner, event);
       if (dr != null) {
         dr.removeClearCountReference();
       }
     }
   }

   @Override
   public void txApplyInvalidate(Object key, Object newValue, boolean didDestroy, TransactionId txId,
       TXRmtEvent txEvent, boolean localOp, EventID eventId, Object aCallbackArgument,
       List<EntryEventImpl> pendingCallbacks, FilterRoutingInfo filterRoutingInfo,
       ClientProxyMembershipID bridgeContext, TXEntryState txEntryState, VersionTag versionTag,
       long tailKey) {
     assert pendingCallbacks != null;
     // boolean didInvalidate = false;
     final LocalRegion owner = _getOwner();

     @Released
     EntryEventImpl callbackEvent = null;
     boolean forceNewEntry = !owner.isInitialized() && owner.isAllEvents();

     final boolean hasRemoteOrigin = !((TXId) txId).getMemberId().equals(owner.getMyId());
     DiskRegion dr = owner.getDiskRegion();
     // Fix for Bug #44431. We do NOT want to update the region and wait
     // later for index INIT as region.clear() can cause inconsistency if
     // happened in parallel as it also does index INIT.
     IndexManager oqlIndexManager = owner.getIndexManager();
     if (oqlIndexManager != null) {
       oqlIndexManager.waitForIndexInit();
     }
     final boolean locked = owner.lockWhenRegionIsInitializing();
     try {
       if (forceNewEntry) {
         boolean opCompleted = false;
         RegionEntry newRe = getEntryFactory().createEntry(owner, key, Token.REMOVED_PHASE1);
         synchronized (newRe) {
           try {
             RegionEntry oldRe = putEntryIfAbsent(key, newRe);
             while (!opCompleted && oldRe != null) {
               synchronized (oldRe) {
                 if (oldRe.isRemovedPhase2()) {
                   owner.getCachePerfStats().incRetries();
                   getEntryMap().remove(key, oldRe);
                   oldRe = putEntryIfAbsent(key, newRe);
                 } else {
                   opCompleted = true;
                   final boolean oldWasTombstone = oldRe.isTombstone();
                   final int oldSize = owner.calculateRegionEntryValueSize(oldRe);
                   Object oldValue = oldRe.getValueInVM(owner); // OFFHEAP eei
                   // Create an entry event only if the calling context is
                   // a receipt of a TXCommitMessage AND there are callbacks
                   // installed
                   // for this region
                   boolean invokeCallbacks = shouldInvokeCallbacks(owner, owner.isInitialized());
                   boolean callbackEventInPending = false;
                   callbackEvent = txCallbackEventFactory.createCallbackEvent(owner,
                       localOp ? Operation.LOCAL_INVALIDATE : Operation.INVALIDATE, key, newValue,
                       txId, txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                       txEntryState, versionTag, tailKey);
                   try {
                     callbackEvent.setRegionEntry(oldRe);
                     callbackEvent.setOldValue(oldValue);
                     if (logger.isDebugEnabled()) {
                       logger.debug("txApplyInvalidate callbackEvent={}", callbackEvent);
                     }

                     txRemoveOldIndexEntry(Operation.INVALIDATE, oldRe);
                     if (didDestroy) {
                       oldRe.txDidDestroy(owner.cacheTimeMillis());
                     }
                     if (txEvent != null) {
                       txEvent.addInvalidate(owner, oldRe, oldRe.getKey(), newValue,
                           aCallbackArgument);
                     }
                     oldRe.setValueResultOfSearch(false);
                     processAndGenerateTXVersionTag(callbackEvent, oldRe, txEntryState);
                     boolean clearOccured = false;
                     try {
                       oldRe.setValue(owner, oldRe.prepareValueForCache(owner, newValue, true));
                       EntryLogger.logTXInvalidate(_getOwnerObject(), key);
                       owner.updateSizeOnPut(key, oldSize, 0);
                       if (oldWasTombstone) {
                         owner.unscheduleTombstone(oldRe);
                       }
                     } catch (RegionClearedException rce) {
                       clearOccured = true;
                     }
                     owner.txApplyInvalidatePart2(oldRe, oldRe.getKey(), didDestroy, true);
                     // didInvalidate = true;
                     if (invokeCallbacks) {
                       switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                       pendingCallbacks.add(callbackEvent);
                       callbackEventInPending = true;
                     }
                     if (!clearOccured) {
                       lruEntryUpdate(oldRe);
                     }
                     if (shouldPerformConcurrencyChecks(owner, callbackEvent)
                         && txEntryState != null) {
                       txEntryState.setVersionTag(callbackEvent.getVersionTag());
                     }
                   } finally {
                     if (!callbackEventInPending)
                       releaseEvent(callbackEvent);
                   }
                 }
               }
             }
             if (!opCompleted) {
               boolean invokeCallbacks = shouldInvokeCallbacks(owner, owner.isInitialized());
               boolean callbackEventInPending = false;
               callbackEvent = txCallbackEventFactory.createCallbackEvent(owner,
                   localOp ? Operation.LOCAL_INVALIDATE : Operation.INVALIDATE, key, newValue, txId,
                   txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                   txEntryState, versionTag, tailKey);
               try {
                 callbackEvent.setRegionEntry(newRe);
                 txRemoveOldIndexEntry(Operation.INVALIDATE, newRe);
                 newRe.setValueResultOfSearch(false);
                 boolean clearOccured = false;
                 try {
                   processAndGenerateTXVersionTag(callbackEvent, newRe, txEntryState);
                   newRe.setValue(owner, newRe.prepareValueForCache(owner, newValue, true));
                   EntryLogger.logTXInvalidate(_getOwnerObject(), key);
                   owner.updateSizeOnCreate(newRe.getKey(), 0);// we are putting in a new invalidated
                                                               // entry
                 } catch (RegionClearedException rce) {
                   clearOccured = true;
                 }
                 owner.txApplyInvalidatePart2(newRe, newRe.getKey(), didDestroy, true);

                 if (invokeCallbacks) {
                   switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                   pendingCallbacks.add(callbackEvent);
                   callbackEventInPending = true;
                 }
                 opCompleted = true;
                 if (!clearOccured) {
                   lruEntryCreate(newRe);
                   incEntryCount(1);
                 }
                 if (shouldPerformConcurrencyChecks(owner, callbackEvent) && txEntryState != null) {
                   txEntryState.setVersionTag(callbackEvent.getVersionTag());
                 }
               } finally {
                 if (!callbackEventInPending)
                   releaseEvent(callbackEvent);
               }
             }
           } finally {
             if (!opCompleted) {
               removeEntry(key, newRe, false);
             }
           }
         }
       } else { /* !forceNewEntry */
         RegionEntry re = getEntry(key);
         if (re != null) {
           synchronized (re) {
             // Fix GEODE-3204, do not invalidate the region entry if it is a removed token
             if (!Token.isRemoved(re.getValueAsToken())) {
               final int oldSize = owner.calculateRegionEntryValueSize(re);
               Object oldValue = re.getValueInVM(owner); // OFFHEAP eei
               // Create an entry event only if the calling context is
               // a receipt of a TXCommitMessage AND there are callbacks
               // installed
               // for this region
               boolean invokeCallbacks = shouldInvokeCallbacks(owner, owner.isInitialized());
               boolean callbackEventInPending = false;
               callbackEvent = txCallbackEventFactory.createCallbackEvent(owner,
                   localOp ? Operation.LOCAL_INVALIDATE : Operation.INVALIDATE, key, newValue, txId,
                   txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext,
                   txEntryState, versionTag, tailKey);
               try {
                 callbackEvent.setRegionEntry(re);
                 callbackEvent.setOldValue(oldValue);
                 txRemoveOldIndexEntry(Operation.INVALIDATE, re);
                 if (didDestroy) {
                   re.txDidDestroy(owner.cacheTimeMillis());
                 }
                 if (txEvent != null) {
                   txEvent.addInvalidate(owner, re, re.getKey(), newValue, aCallbackArgument);
                 }
                 re.setValueResultOfSearch(false);
                 processAndGenerateTXVersionTag(callbackEvent, re, txEntryState);
                 boolean clearOccured = false;
                 try {
                   re.setValue(owner, re.prepareValueForCache(owner, newValue, true));
                   EntryLogger.logTXInvalidate(_getOwnerObject(), key);
                   owner.updateSizeOnPut(key, oldSize, 0);
                 } catch (RegionClearedException rce) {
                   clearOccured = true;
                 }
                 owner.txApplyInvalidatePart2(re, re.getKey(), didDestroy, true);
                 // didInvalidate = true;
                 if (invokeCallbacks) {
                   switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
                   pendingCallbacks.add(callbackEvent);
                   callbackEventInPending = true;
                 }
                 if (!clearOccured) {
                   lruEntryUpdate(re);
                 }
                 if (shouldPerformConcurrencyChecks(owner, callbackEvent) && txEntryState != null) {
                   txEntryState.setVersionTag(callbackEvent.getVersionTag());
                 }
               } finally {
                 if (!callbackEventInPending)
                   releaseEvent(callbackEvent);
               }
               return;
             }
           }
         }
         { // re == null or region entry is removed token.
           // Fix bug#43594
           // In cases where bucket region is re-created, it may so happen
           // that the invalidate is already applied on the Initial image
           // provider, thus causing region entry to be absent.
           // Notify clients with client events.
           boolean callbackEventInPending = false;
           callbackEvent = txCallbackEventFactory.createCallbackEvent(owner,
               localOp ? Operation.LOCAL_INVALIDATE : Operation.INVALIDATE, key, newValue, txId,
               txEvent, eventId, aCallbackArgument, filterRoutingInfo, bridgeContext, txEntryState,
               versionTag, tailKey);
           try {
             switchEventOwnerAndOriginRemote(callbackEvent, hasRemoteOrigin);
             pendingCallbacks.add(callbackEvent);
             callbackEventInPending = true;
           } finally {
             if (!callbackEventInPending)
               releaseEvent(callbackEvent);
           }
         }
       }
     } catch (DiskAccessException dae) {
       owner.handleDiskAccessException(dae);
       throw dae;
     } finally {
       if (locked) {
         owner.unlockWhenRegionIsInitializing();
       }
       if (oqlIndexManager != null) {
         oqlIndexManager.countDownIndexUpdaters();
       }
     }
   }

   /**
    * This code needs to be evaluated. It was added quickly to help PR persistence not to consume as
    * much memory.
    */
   @Override
   public void evictValue(Object key) {
     final LocalRegion owner = _getOwner();
     RegionEntry re = getEntry(key);
     if (re != null) {
       synchronized (re) {
         if (!re.isValueNull()) {
           re.setValueToNull();
           owner.getDiskRegion().incNumEntriesInVM(-1L);
           owner.getDiskRegion().incNumOverflowOnDisk(1L);
           if (owner instanceof BucketRegion) {
             ((BucketRegion) owner).incNumEntriesInVM(-1L);
             ((BucketRegion) owner).incNumOverflowOnDisk(1L);
           }
         }
       }
     }
   }

   /*
    * returns null if the operation fails
    */
   @Override
   public RegionEntry basicPut(EntryEventImpl event, final long unused, final boolean ifNew,
       final boolean ifOld, final Object expectedOldValue, // only non-null if ifOld
       final boolean requireOldValue, final boolean overwriteDestroyed)
       throws CacheWriterException, TimeoutException {

     final RegionMapPut regionMapPut =
         new RegionMapPut(this, _getOwner(), this, entryEventSerialization, event, ifNew, ifOld,
             overwriteDestroyed, requireOldValue, expectedOldValue);

     return regionMapPut.put();
   }

   @Override
   public void runWhileEvictionDisabled(Runnable r) {
     final boolean disabled = disableLruUpdateCallback();
     try {
       r.run();
     } finally {
       if (disabled) {
         enableLruUpdateCallback();
       }
     }
   }

   @Override
   public void txApplyPut(Operation putOp, Object key, Object nv, boolean didDestroy,
       TransactionId txId, TXRmtEvent txEvent, EventID eventId, Object aCallbackArgument,
       List<EntryEventImpl> pendingCallbacks, FilterRoutingInfo filterRoutingInfo,
       ClientProxyMembershipID bridgeContext, TXEntryState txEntryState, VersionTag versionTag,
       long tailKey) {
     assert pendingCallbacks != null;
     final LocalRegion owner = _getOwner();
     @Released
     final EntryEventImpl callbackEvent =
         createTransactionCallbackEvent(owner, putOp, key, nv, txId, txEvent, eventId,
             aCallbackArgument, filterRoutingInfo, bridgeContext, txEntryState, versionTag, tailKey);
     if (owner.isUsedForPartitionedRegionBucket()) {
       callbackEvent.makeSerializedNewValue();
       txHandleWANEvent(owner, callbackEvent, txEntryState);
     }

     RegionMapCommitPut commitPut = new RegionMapCommitPut(this, owner, callbackEvent, putOp,
         didDestroy, txId, txEvent, pendingCallbacks, txEntryState);
     commitPut.put();
   }

   private void txHandleWANEvent(final LocalRegion owner, EntryEventImpl callbackEvent,
       TXEntryState txEntryState) {
     owner.handleWANEvent(callbackEvent);
     if (txEntryState != null) {
       txEntryState.setTailKey(callbackEvent.getTailKey());
     }
   }

   /**
    * called from txApply* methods to process and generate versionTags.
    */
   @Override
   public void processAndGenerateTXVersionTag(EntryEventImpl callbackEvent, RegionEntry re,
       TXEntryState txEntryState) {
     final LocalRegion owner = _getOwner();
     if (shouldPerformConcurrencyChecks(owner, callbackEvent)) {
       try {
         if (txEntryState != null && txEntryState.getRemoteVersionTag() != null) {
           // to generate a version based on a remote VersionTag, we will
           // have to put the remote versionTag in the regionEntry
           VersionTag remoteTag = txEntryState.getRemoteVersionTag();
           if (re instanceof VersionStamp) {
             VersionStamp stamp = (VersionStamp) re;
             stamp.setVersions(remoteTag);
           }
         }
         processVersionTag(re, callbackEvent);
       } catch (ConcurrentCacheModificationException ignore) {
         // ignore this exception, however invoke callbacks for this operation
       }

       // For distributed transactions, stuff the next region version generated
       // in phase-1 commit into the callbackEvent so that ARE.generateVersionTag can later
       // just apply it and not regenerate it in phase-2 commit
       if (txEntryState != null
           && txEntryState.getDistTxEntryStates() != null) {
         callbackEvent.setNextRegionVersion(txEntryState.getDistTxEntryStates().getRegionVersion());
       }

       // callbackEvent.setNextRegionVersion(txEntryState.getNextRegionVersion());
       owner.generateAndSetVersionTag(callbackEvent, re);
     }
   }

   /**
    * Checks for concurrency checks enabled on Region and that callbackEvent is not null.
    */
   private boolean shouldPerformConcurrencyChecks(LocalRegion owner, EntryEventImpl callbackEvent) {
     return owner.getConcurrencyChecksEnabled() && callbackEvent != null;
   }

   /**
    * Removing the existing indexed value requires the current value in the cache, that is the one
    * prior to applying the operation.
    *
    * @param entry the RegionEntry that contains the value prior to applying the op
    */
   @Override
   public void txRemoveOldIndexEntry(Operation op, RegionEntry entry) {
     if ((op.isUpdate() && !entry.isInvalid()) || op.isInvalidate() || op.isDestroy()) {
       IndexManager idxManager = _getOwner().getIndexManager();
       if (idxManager != null) {
         try {
           idxManager.updateIndexes(entry, IndexManager.REMOVE_ENTRY,
               op.isUpdate() ? IndexProtocol.BEFORE_UPDATE_OP : IndexProtocol.OTHER_OP);
         } catch (QueryException e) {
           throw new IndexMaintenanceException(e);
         }
       }
     }
   }

   public void dumpMap() {
     logger.info("dump of concurrent map of size {} for region {}", getEntryMap().size(),
         this._getOwner());
     for (Iterator it = getEntryMap().values().iterator(); it.hasNext();) {
       logger.info("dumpMap:" + it.next().toString());
     }
   }

   EntryEventImpl createTransactionCallbackEvent(final LocalRegion re, Operation op, Object key,
       Object newValue, TransactionId txId, TXRmtEvent txEvent, EventID eventId,
       Object aCallbackArgument, FilterRoutingInfo filterRoutingInfo,
       ClientProxyMembershipID bridgeContext, TXEntryState txEntryState, VersionTag versionTag,
       long tailKey) {
     return txCallbackEventFactory
         .createCallbackEvent(re, op, key, newValue, txId, txEvent, eventId, aCallbackArgument,
             filterRoutingInfo, bridgeContext, txEntryState, versionTag, tailKey);
   }

   @Override
   public void writeSyncIfPresent(Object key, Runnable runner) {
     RegionEntry re = getEntry(key);
     if (re != null) {
       final boolean disabled = disableLruUpdateCallback();
       try {
         synchronized (re) {
           if (!re.isRemoved()) {
             runner.run();
           }
         }
       } finally {
         if (disabled) {
           enableLruUpdateCallback();
         }
         try {
           lruUpdateCallback();
         } catch (DiskAccessException dae) {
           this._getOwner().handleDiskAccessException(dae);
           throw dae;
         }
       }
     }
   }

   @Override
   public void removeIfDestroyed(Object key) {
     LocalRegion owner = _getOwner();
     // boolean makeTombstones = owner.concurrencyChecksEnabled;
     DiskRegion dr = owner.getDiskRegion();
     RegionEntry re = getEntry(key);
     if (re != null) {
       if (re.isDestroyed()) {
         synchronized (re) {
           if (re.isDestroyed()) {
             // [bruce] destroyed entries aren't in the LRU clock, so they can't be retained here
             // if (makeTombstones) {
             // re.makeTombstone(owner, re.getVersionStamp().asVersionTag());
             // } else {
             re.removePhase2();
             removeEntry(key, re, true);
           }
         }
       }
     }
     // }
   }

   /** get version-generation permission from the region's version vector */
   @Override
   public void lockForCacheModification(InternalRegion owner, EntryEventImpl event) {
     boolean lockedByBulkOp = event.isBulkOpInProgress() && owner.getDataPolicy().withReplication();

     if (armLockTestHook != null) {
       armLockTestHook.beforeLock(owner, event);
     }

     if (!event.isOriginRemote() && !lockedByBulkOp && !owner.hasServerProxy()) {
       RegionVersionVector vector = owner.getVersionVector();
       if (vector != null) {
         vector.lockForCacheModification();
       }
     }

     if (armLockTestHook != null) {
       armLockTestHook.afterLock(owner, event);
     }
   }

   /** release version-generation permission from the region's version vector */
   @Override
   public void releaseCacheModificationLock(InternalRegion owner, EntryEventImpl event) {
     boolean lockedByBulkOp = event.isBulkOpInProgress() && owner.getDataPolicy().withReplication();

     if (armLockTestHook != null)
       armLockTestHook.beforeRelease(owner, event);

     if (!event.isOriginRemote() && !lockedByBulkOp && !owner.hasServerProxy()) {
       RegionVersionVector vector = owner.getVersionVector();
       if (vector != null) {
         vector.releaseCacheModificationLock();
       }
     }

     if (armLockTestHook != null)
       armLockTestHook.afterRelease(owner, event);

   }

   @Override
   public void lockRegionForAtomicTX(InternalRegion r) {
     if (armLockTestHook != null)
       armLockTestHook.beforeLock(r, null);

     RegionVersionVector vector = r.getVersionVector();
     if (vector != null) {
       vector.lockForCacheModification();
     }

     if (armLockTestHook != null)
       armLockTestHook.afterLock(r, null);
   }

   @Override
   public void unlockRegionForAtomicTX(InternalRegion r) {
     if (armLockTestHook != null)
       armLockTestHook.beforeRelease(r, null);

     RegionVersionVector vector = r.getVersionVector();
     if (vector != null) {
       vector.releaseCacheModificationLock();
     }

     if (armLockTestHook != null)
       armLockTestHook.afterRelease(r, null);
   }

   /**
    * for testing race conditions between threads trying to apply ops to the same entry
    *
    * @param entry the entry to attempt to add to the system
    */
   protected RegionEntry putEntryIfAbsentForTest(RegionEntry entry) {
     return (RegionEntry) putEntryIfAbsent(entry.getKey(), entry);
   }

   @Override
   public boolean isTombstoneNotNeeded(RegionEntry re, int destroyedVersion) {
     // no need for synchronization - stale values are okay here
     // TODO this looks like a problem for regionEntry pooling
     if (getEntry(re.getKey()) != re) {
       // region entry was either removed (null)
       // or changed to a different region entry.
       // In either case the old tombstone is no longer needed.
       return true;
     }
     if (!re.isTombstone()) {
       // if the region entry no longer contains a tombstone
       // then the old tombstone is no longer needed
       return true;
     }
     VersionStamp<?> vs = re.getVersionStamp();
     if (vs == null) {
       // if we have no VersionStamp why were we even added as a tombstone?
       // We used to see an NPE here. See bug 52092.
       logger.error(
           "Unexpected RegionEntry scheduled as tombstone: re.getClass {} destroyedVersion {}",
           re.getClass(), destroyedVersion);
       return true;
     }
     if (vs.getEntryVersion() != destroyedVersion) {
       // the version changed so old tombstone no longer needed
       return true;
     }
     // region entry still has the same tombstone so we need to keep it.
     return false;
   }

   /** removes a tombstone that has expired locally */
   @Override
   public boolean removeTombstone(RegionEntry re, VersionHolder version, boolean isEviction,
       boolean isScheduledTombstone) {
     boolean result = false;
     int destroyedVersion = version.getEntryVersion();

     synchronized (this._getOwner().getSizeGuard()) { // do this sync first; see bug 51985
       synchronized (re) {
         int entryVersion = re.getVersionStamp().getEntryVersion();
         if (!re.isTombstone() || entryVersion > destroyedVersion) {
           if (logger.isTraceEnabled(LogMarker.TOMBSTONE_COUNT_VERBOSE)) {
             logger.trace(LogMarker.TOMBSTONE_COUNT_VERBOSE,
                 "tombstone for {} was resurrected with v{}; destroyed version was v{}; count is {}; entryMap size is {}",
                 re.getKey(), re.getVersionStamp().getEntryVersion(), destroyedVersion,
                 this._getOwner().getTombstoneCount(), size());
           }
         } else {
           if (logger.isTraceEnabled(LogMarker.TOMBSTONE_COUNT_VERBOSE)) {
             if (entryVersion == destroyedVersion) {
               // logging this can put tremendous pressure on the log writer in tests
               // that "wait for silence"
               logger.trace(LogMarker.TOMBSTONE_COUNT_VERBOSE,
                   "removing tombstone for {} with v{} rv{}; count is {}", re.getKey(),
                   destroyedVersion, version.getRegionVersion(),
                   (this._getOwner().getTombstoneCount() - 1));
             } else {
               logger.trace(LogMarker.TOMBSTONE_COUNT_VERBOSE,
                   "removing entry (v{}) that is older than an expiring tombstone (v{} rv{}) for {}",
                   entryVersion, destroyedVersion, version.getRegionVersion(), re.getKey());
             }
           }
           try {
             re.setValue(_getOwner(), Token.REMOVED_PHASE2);
             if (removeTombstone(re)) {
               _getOwner().cancelExpiryTask(re);
               result = true;
               incEntryCount(-1);
               // Bug 51118: When the method is called by tombstoneGC thread, current 're' is an
               // expired tombstone. Then we detected an destroyed (due to overwritingOldTombstone()
               // returns true earlier) tombstone with bigger entry version, it's safe to delete
               // current tombstone 're' and adjust the tombstone count.
               // lruEntryDestroy(re); // tombstones are invisible to LRU
               if (isScheduledTombstone) {
                 _getOwner().incTombstoneCount(-1);
               }
               RegionVersionVector vector = _getOwner().getVersionVector();
               if (vector != null) {
                 vector.recordGCVersion(version.getMemberID(), version.getRegionVersion());
               }
             }
           } catch (RegionClearedException e) {
             // if the region has been cleared we don't need to remove the tombstone
           } catch (RegionDestroyedException e) {
             // if the region has been destroyed, the tombstone is already
             // gone. Catch an exception to avoid an error from the GC thread.
           }
         }
       }
     }
     return result;
   }

   private boolean removeTombstone(RegionEntry re) {
     return getEntryMap().remove(re.getKey(), re);
   }

   // method used for debugging tombstone count issues
   public boolean verifyTombstoneCount(AtomicInteger numTombstones) {
     int deadEntries = 0;
     try {
       for (Iterator it = getEntryMap().values().iterator(); it.hasNext();) {
         RegionEntry re = (RegionEntry) it.next();
         if (re.isTombstone()) {
           deadEntries++;
         }
       }
       if (deadEntries != numTombstones.get()) {
         if (logger.isDebugEnabled()) {
           logger.debug("tombstone count ({}) does not match actual number of tombstones ({})",
               numTombstones, deadEntries, new Exception());
         }
         return false;
       } else {
         if (logger.isDebugEnabled()) {
           logger.debug("tombstone count verified");
         }
       }
     } catch (Exception e) {
       // ignore
     }
     return true;
   }

   @Override
   public int getEntryOverhead() {
     return (int) ReflectionSingleObjectSizer.sizeof(getEntryFactory().getEntryClass());
   }

   private ARMLockTestHook armLockTestHook;

   @Override
   public ARMLockTestHook getARMLockTestHook() {
     return armLockTestHook;
   }

   public void setARMLockTestHook(ARMLockTestHook theHook) {
     armLockTestHook = theHook;
   }
 }