/*
 * 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.OffHeapClearRequired;
import org.apache.geode.internal.offheap.OffHeapHelper;
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) {
    entryFactory = f;
  }

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

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

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

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

  boolean _isOwnerALocalRegion() {
    return owner instanceof LocalRegion;
  }

  Object _getOwnerObject() {
    return 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;
    }
    return !re.isRemoved();
  }

  @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 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 = 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<>();

    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
    synchronized (lr.getSizeGuard()) {
      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 (OffHeapClearRequired.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<>();
    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);
                  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) {
                    if (!oldIsDestroyedOrRemoved) {
                      owner.updateSizeOnRemove(key, oldSize);
                    }
                    owner.scheduleTombstone(oldRe, entryVersion);
                    if (!oldIsTombstone) {
                      lruEntryDestroy(oldRe);
                    }
                  } else {
                    int newSize = owner.calculateRegionEntryValueSize(oldRe);
                    if (!oldIsTombstone) {
                      owner.updateSizeOnPut(key, oldSize, newSize);
                      lruEntryUpdate(oldRe);
                    } else {
                      owner.updateSizeOnCreate(key, newSize);
                      lruEntryCreate(oldRe);
                    }
                    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);
