/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache;

import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_NEW_VALUE;
import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_OLD_VALUE;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.function.Function;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CopyHelper;
import org.apache.geode.DataSerializer;
import org.apache.geode.Delta;
import org.apache.geode.DeltaSerializationException;
import org.apache.geode.GemFireIOException;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.SerializationException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.SerializedCacheValue;
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.cache.query.internal.index.IndexUtils;
import org.apache.geode.cache.util.TimestampedEntryEvent;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.DSFIDFactory;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.Sendable;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.entries.OffHeapRegionEntry;
import org.apache.geode.internal.cache.partitioned.PartitionMessage;
import org.apache.geode.internal.cache.partitioned.PutMessage;
import org.apache.geode.internal.cache.tier.sockets.CacheServerHelper;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tx.DistTxKeyInfo;
import org.apache.geode.internal.cache.tx.RemoteOperationMessage;
import org.apache.geode.internal.cache.tx.RemotePutMessage;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.cache.wan.GatewaySenderEventCallbackArgument;
import org.apache.geode.internal.lang.StringUtils;
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.ReferenceCountHelper;
import org.apache.geode.internal.offheap.Releasable;
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.offheap.annotations.Unretained;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.size.Sizeable;
import org.apache.geode.internal.util.ArrayUtils;
import org.apache.geode.internal.util.BlobHelper;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.internal.PeerTypeRegistration;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implementation of an entry event
 *
 * must be public for DataSerializableFixedID
 */
public class EntryEventImpl implements InternalEntryEvent, InternalCacheEvent,
    DataSerializableFixedID, EntryOperation, Releasable {
  private static final Logger logger = LogService.getLogger();

  // PACKAGE FIELDS //
  private transient InternalRegion region;

  private transient RegionEntry re;

  protected KeyInfo keyInfo;

  /** the event's id. Scoped by distributedMember. */
  protected EventID eventID;

  private Object newValue = null;

  /**
   * If we ever serialize the new value then it should be stored in this field in case we need the
   * serialized form again later. This was added to fix bug 43781. Note that we also have the
   * "newValueBytes" field. But it is only non-null if setSerializedNewValue was called.
   */
  private byte[] cachedSerializedNewValue = null;

  @Retained(ENTRY_EVENT_OLD_VALUE)
  private Object oldValue = null;

  protected short eventFlags = 0x0000;

  protected TXId txId = null;

  protected Operation op;

  /* To store the operation/modification type */
  private transient EnumListenerEvent eventType;

  /**
   * This field will be null unless this event is used for a putAll operation.
   *
   * @since GemFire 5.0
   */
  protected transient DistributedPutAllOperation putAllOp;

  /**
   * This field will be null unless this event is used for a removeAll operation.
   *
   * @since GemFire 8.1
   */
  protected transient DistributedRemoveAllOperation removeAllOp;

  /**
   * The member that originated this event
   *
   * @since GemFire 5.0
   */
  protected DistributedMember distributedMember;

  /**
   * transient storage for the message that caused the event
   */
  transient DistributionMessage causedByMessage;

  /**
   * The originating membershipId of this event.
   *
   * @since GemFire 5.1
   */
  protected ClientProxyMembershipID context = null;

  /**
   * this holds the bytes representing the change in value effected by this event. It is used when
   * the value implements the Delta interface.
   */
  private byte[] deltaBytes = null;

  /** routing information for cache clients for this event */
  private FilterInfo filterInfo;

  /** new value stored in serialized form */
  protected byte[] newValueBytes;

  /** old value stored in serialized form */
  private byte[] oldValueBytes;

  /** version tag for concurrency checks */
  protected VersionTag versionTag;

  /** boolean to indicate that the RegionEntry for this event has been evicted */
  private transient boolean isEvicted = false;

  private transient boolean isPendingSecondaryExpireDestroy = false;

  private transient boolean hasRetried = false;

  public static final Object SUSPECT_TOKEN = new Object();

  public EntryEventImpl() {
    offHeapLock = null;
  }

  /**
   * Reads the contents of this message from the given input.
   */
  @Override
  public void fromData(DataInput in,
      DeserializationContext context) throws IOException, ClassNotFoundException {
    eventID = context.getDeserializer().readObject(in);
    Object key = context.getDeserializer().readObject(in);
    Object value = context.getDeserializer().readObject(in);
    keyInfo = new KeyInfo(key, value, null);
    op = Operation.fromOrdinal(in.readByte());
    eventFlags = in.readShort();
    keyInfo.setCallbackArg(context.getDeserializer().readObject(in));
    txId = context.getDeserializer().readObject(in);

    if (in.readBoolean()) { // isDelta
      assert false : "isDelta should never be true";
    } else {
      // OFFHEAP Currently values are never deserialized to off heap memory. If that changes then
      // this code needs to change.
      if (in.readBoolean()) { // newValueSerialized
        newValueBytes = DataSerializer.readByteArray(in);
        cachedSerializedNewValue = newValueBytes;
        newValue = null; // set later in generateNewValueFromBytesIfNeeded
      } else {
        newValueBytes = null;
        cachedSerializedNewValue = null;
        newValue = context.getDeserializer().readObject(in);
      }
    }

    // OFFHEAP Currently values are never deserialized to off heap memory. If that changes then this
    // code needs to change.
    if (in.readBoolean()) { // oldValueSerialized
      oldValueBytes = DataSerializer.readByteArray(in);
      oldValue = null; // set later in basicGetOldValue
    } else {
      oldValueBytes = null;
      oldValue = context.getDeserializer().readObject(in);
    }
    distributedMember = DSFIDFactory.readInternalDistributedMember(in);
    this.context = ClientProxyMembershipID.readCanonicalized(in);
    tailKey = DataSerializer.readLong(in);
  }

  @Retained
  protected EntryEventImpl(InternalRegion region, Operation op, Object key, boolean originRemote,
      DistributedMember distributedMember, boolean generateCallbacks, boolean fromRILocalDestroy) {
    this.region = region;
    InternalDistributedSystem ds =
        (InternalDistributedSystem) region.getCache().getDistributedSystem();
    if (ds.getOffHeapStore() != null) {
      offHeapLock = new Object();
    } else {
      offHeapLock = null;
    }
    this.op = op;
    keyInfo = region.getKeyInfo(key);
    setOriginRemote(originRemote);
    setGenerateCallbacks(generateCallbacks);
    this.distributedMember = distributedMember;
    setFromRILocalDestroy(fromRILocalDestroy);
  }

  /**
   * Doesn't specify oldValue as this will be filled in later as part of an operation on the region,
   * or lets it default to null.
   */
  @Retained
  protected EntryEventImpl(final InternalRegion region, Operation op, Object key,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object newVal, Object callbackArgument, boolean originRemote,
      DistributedMember distributedMember, boolean generateCallbacks, boolean initializeId) {
    this.region = region;
    InternalDistributedSystem ds =
        (InternalDistributedSystem) region.getCache().getDistributedSystem();
    if (ds.getOffHeapStore() != null) {
      offHeapLock = new Object();
    } else {
      offHeapLock = null;
    }
    this.op = op;
    keyInfo = region.getKeyInfo(key, newVal, callbackArgument);

    if (!Token.isInvalid(newVal)) {
      basicSetNewValue(newVal, false);
    }

    txId = region.getTXId();
    /*
     * this might set txId for events done from a thread that has a tx even though the op is non-tx.
     * For example region ops.
     */
    if (newVal == Token.LOCAL_INVALID) {
      setLocalInvalid(true);
    }
    setOriginRemote(originRemote);
    setGenerateCallbacks(generateCallbacks);
    this.distributedMember = distributedMember;
  }

  /**
   * Called by BridgeEntryEventImpl to use existing EventID
   */
  @Retained
  protected EntryEventImpl(InternalRegion region, Operation op, Object key,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object newValue, Object callbackArgument,
      boolean originRemote, DistributedMember distributedMember, boolean generateCallbacks,
      EventID eventID) {
    this(region, op, key, newValue, callbackArgument, originRemote, distributedMember,
        generateCallbacks, true /* initializeId */);
    Assert.assertTrue(eventID != null || !(region instanceof PartitionedRegion));
    setEventId(eventID);
  }

  /**
   * create an entry event from another entry event
   */
  @Retained
  public EntryEventImpl(
      @Retained({ENTRY_EVENT_NEW_VALUE, ENTRY_EVENT_OLD_VALUE}) EntryEventImpl other) {
    this(other, true);
  }

  @Retained
  public EntryEventImpl(
      @Retained({ENTRY_EVENT_NEW_VALUE, ENTRY_EVENT_OLD_VALUE}) EntryEventImpl other,
      boolean setOldValue) {
    setRegion(other.getRegion());
    if (other.offHeapLock != null) {
      offHeapLock = new Object();
    } else {
      offHeapLock = null;
    }

    eventID = other.eventID;
    basicSetNewValue(other.basicGetNewValue(), false);
    newValueBytes = other.newValueBytes;
    cachedSerializedNewValue = other.cachedSerializedNewValue;
    re = other.re;
    if (setOldValue) {
      retainAndSetOldValue(other.basicGetOldValue());
      oldValueBytes = other.oldValueBytes;
    }
    eventFlags = other.eventFlags;
    setEventFlag(EventFlags.FLAG_CALLBACKS_INVOKED, false);
    txId = other.txId;
    op = other.op;
    distributedMember = other.distributedMember;
    filterInfo = other.filterInfo;
    keyInfo =
        other.getKeyInfo().isDistKeyInfo() ? new DistTxKeyInfo((DistTxKeyInfo) other.getKeyInfo())
            : new KeyInfo(other.getKeyInfo());
    if (other.getRawCallbackArgument() instanceof GatewaySenderEventCallbackArgument) {
      keyInfo.setCallbackArg((new GatewaySenderEventCallbackArgument(
          (GatewaySenderEventCallbackArgument) other.getRawCallbackArgument())));
    }
    context = other.context;
    deltaBytes = other.deltaBytes;
    tailKey = other.tailKey;
    versionTag = other.versionTag;
    // set possible duplicate
    setPossibleDuplicate(other.isPossibleDuplicate());
  }

  @Retained
  public EntryEventImpl(Object key2, boolean isOffHeap) {
    keyInfo = new KeyInfo(key2, null, null);
    if (isOffHeap) {
      offHeapLock = new Object();
    } else {
      offHeapLock = null;
    }
  }

  /**
   * Creates and returns an EntryEventImpl. Generates and assigns a bucket id to the EntryEventImpl
   * if the region parameter is a PartitionedRegion.
   */
  @Retained
  public static EntryEventImpl create(InternalRegion region, Operation op, Object key,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object newValue, Object callbackArgument,
      boolean originRemote, DistributedMember distributedMember) {
    return create(region, op, key, newValue, callbackArgument, originRemote, distributedMember,
        true, true);
  }

  /**
   * Creates and returns an EntryEventImpl. Generates and assigns a bucket id to the EntryEventImpl
   * if the region parameter is a PartitionedRegion.
   */
  @Retained
  public static EntryEventImpl create(InternalRegion region, Operation op, Object key,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object newValue, Object callbackArgument,
      boolean originRemote, DistributedMember distributedMember, boolean generateCallbacks) {
    return create(region, op, key, newValue, callbackArgument, originRemote, distributedMember,
        generateCallbacks, true);
  }

  /**
   * Creates and returns an EntryEventImpl. Generates and assigns a bucket id to the EntryEventImpl
   * if the region parameter is a PartitionedRegion.
   *
   * Called by BridgeEntryEventImpl to use existing EventID
   */
  @Retained
  public static EntryEventImpl create(InternalRegion region, Operation op, Object key,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object newValue, Object callbackArgument,
      boolean originRemote, DistributedMember distributedMember, boolean generateCallbacks,
      EventID eventID) {
    return new EntryEventImpl(region, op, key, newValue, callbackArgument, originRemote,
        distributedMember, generateCallbacks, eventID);
  }

  /**
   * Creates and returns an EntryEventImpl. Generates and assigns a bucket id to the EntryEventImpl
   * if the region parameter is a PartitionedRegion.
   */
  @Retained
  public static EntryEventImpl create(InternalRegion region, Operation op, Object key,
      boolean originRemote, DistributedMember distributedMember, boolean generateCallbacks,
      boolean fromRILocalDestroy) {
    return new EntryEventImpl(region, op, key, originRemote, distributedMember, generateCallbacks,
        fromRILocalDestroy);
  }

  /**
   * Creates and returns an EntryEventImpl. Generates and assigns a bucket id to the EntryEventImpl
   * if the region parameter is a PartitionedRegion.
   *
   * This creator does not specify the oldValue as this will be filled in later as part of an
   * operation on the region, or lets it default to null.
   */
  @Retained
  public static EntryEventImpl create(final InternalRegion region, Operation op, Object key,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object newVal, Object callbackArgument, boolean originRemote,
      DistributedMember distributedMember, boolean generateCallbacks, boolean initializeId) {
    return new EntryEventImpl(region, op, key, newVal, callbackArgument, originRemote,
        distributedMember, generateCallbacks, initializeId);
  }

  /**
   * Creates a PutAllEvent given the distributed operation, the region, and the entry data.
   *
   * @since GemFire 5.0
   */
  @Retained
  static EntryEventImpl createPutAllEvent(DistributedPutAllOperation putAllOp,
      InternalRegion region, Operation entryOp, Object entryKey,
      @Retained(ENTRY_EVENT_NEW_VALUE) Object entryNewValue) {
    @Retained
    EntryEventImpl e;
    if (putAllOp != null) {
      EntryEventImpl event = putAllOp.getBaseEvent();
      if (event.isBridgeEvent()) {
        e = EntryEventImpl.create(region, entryOp, entryKey, entryNewValue,
            event.getRawCallbackArgument(), false, event.distributedMember,
            event.isGenerateCallbacks());
        e.setContext(event.getContext());
      } else {
        e = EntryEventImpl.create(region, entryOp, entryKey, entryNewValue,
            event.getCallbackArgument(), false, region.getMyId(), event.isGenerateCallbacks());
      }
      e.setPossibleDuplicate(event.isPossibleDuplicate());

    } else {
      e = EntryEventImpl.create(region, entryOp, entryKey, entryNewValue, null, false,
          region.getMyId(), true);
    }

    e.putAllOp = putAllOp;
    return e;
  }

  @Retained
  protected static EntryEventImpl createRemoveAllEvent(DistributedRemoveAllOperation op,
      InternalRegion region, Object entryKey) {
    @Retained
    EntryEventImpl e;
    final Operation entryOp = Operation.REMOVEALL_DESTROY;
    if (op != null) {
      EntryEventImpl event = op.getBaseEvent();
      if (event.isBridgeEvent()) {
        e = EntryEventImpl.create(region, entryOp, entryKey, null, event.getRawCallbackArgument(),
            false, event.distributedMember, event.isGenerateCallbacks());
        e.setContext(event.getContext());
      } else {
        e = EntryEventImpl.create(region, entryOp, entryKey, null, event.getCallbackArgument(),
            false, region.getMyId(), event.isGenerateCallbacks());
      }
      e.setPossibleDuplicate(event.isPossibleDuplicate());

    } else {
      e = EntryEventImpl.create(region, entryOp, entryKey, null, null, false, region.getMyId(),
          true);
    }

    e.removeAllOp = op;
    return e;
  }

  public boolean isBulkOpInProgress() {
    return getPutAllOperation() != null || getRemoveAllOperation() != null;
  }

  /** return the putAll operation for this event, if any */
  public DistributedPutAllOperation getPutAllOperation() {
    return putAllOp;
  }

  public DistributedPutAllOperation setPutAllOperation(DistributedPutAllOperation nv) {
    DistributedPutAllOperation result = putAllOp;
    if (nv != null && nv.getBaseEvent() != null) {
      setCallbackArgument(nv.getBaseEvent().getCallbackArgument());
    }
    putAllOp = nv;
    return result;
  }

  public DistributedRemoveAllOperation getRemoveAllOperation() {
    return removeAllOp;
  }

  public DistributedRemoveAllOperation setRemoveAllOperation(DistributedRemoveAllOperation nv) {
    DistributedRemoveAllOperation result = removeAllOp;
    if (nv != null && nv.getBaseEvent() != null) {
      setCallbackArgument(nv.getBaseEvent().getCallbackArgument());
    }
    removeAllOp = nv;
    return result;
  }

  private boolean testEventFlag(short mask) {
    return EventFlags.isSet(eventFlags, mask);
  }

  private void setEventFlag(short mask, boolean on) {
    eventFlags = EventFlags.set(eventFlags, mask, on);
  }

  @Override
  public DistributedMember getDistributedMember() {
    return distributedMember;
  }

  /////////////////////// INTERNAL BOOLEAN SETTERS
  public void setOriginRemote(boolean b) {
    setEventFlag(EventFlags.FLAG_ORIGIN_REMOTE, b);
  }

  public void setLocalInvalid(boolean b) {
    setEventFlag(EventFlags.FLAG_LOCAL_INVALID, b);
  }

  public void setGenerateCallbacks(boolean b) {
    setEventFlag(EventFlags.FLAG_GENERATE_CALLBACKS, b);
  }

  /** set the the flag telling whether callbacks should be invoked for a partitioned region */
  public void setInvokePRCallbacks(boolean b) {
    setEventFlag(EventFlags.FLAG_INVOKE_PR_CALLBACKS, b);
  }

  /** get the flag telling whether callbacks should be invoked for a partitioned region */
  public boolean getInvokePRCallbacks() {
    return testEventFlag(EventFlags.FLAG_INVOKE_PR_CALLBACKS);
  }

  public boolean getInhibitDistribution() {
    return testEventFlag(EventFlags.FLAG_INHIBIT_DISTRIBUTION);
  }

  public void setInhibitDistribution(boolean b) {
    setEventFlag(EventFlags.FLAG_INHIBIT_DISTRIBUTION, b);
  }

  /** was the entry destroyed or missing and allowed to be destroyed again? */
  public boolean getIsRedestroyedEntry() {
    return testEventFlag(EventFlags.FLAG_REDESTROYED_TOMBSTONE);
  }

  public void setIsRedestroyedEntry(boolean b) {
    setEventFlag(EventFlags.FLAG_REDESTROYED_TOMBSTONE, b);
  }

  public void isConcurrencyConflict(boolean b) {
    setEventFlag(EventFlags.FLAG_CONCURRENCY_CONFLICT, b);
  }

  public boolean isConcurrencyConflict() {
    return testEventFlag(EventFlags.FLAG_CONCURRENCY_CONFLICT);
  }

  /** set the DistributionMessage that caused this event */
  public void setCausedByMessage(DistributionMessage msg) {
    causedByMessage = msg;
  }

  /**
   * get the PartitionMessage that caused this event, or null if the event was not caused by a
   * PartitionMessage
   */
  public PartitionMessage getPartitionMessage() {
    if (causedByMessage != null && causedByMessage instanceof PartitionMessage) {
      return (PartitionMessage) causedByMessage;
    }
    return null;
  }

  /**
   * get the RemoteOperationMessage that caused this event, or null if the event was not caused by a
   * RemoteOperationMessage
   */
  public RemoteOperationMessage getRemoteOperationMessage() {
    if (causedByMessage != null && causedByMessage instanceof RemoteOperationMessage) {
      return (RemoteOperationMessage) causedByMessage;
    }
    return null;
  }

  /////////////// BOOLEAN GETTERS
  public boolean isLocalLoad() {
    return op.isLocalLoad();
  }

  public boolean isNetSearch() {
    return op.isNetSearch();
  }

  public boolean isNetLoad() {
    return op.isNetLoad();
  }

  public boolean isDistributed() {
    return op.isDistributed();
  }

  public boolean isExpiration() {
    return op.isExpiration();
  }

  public boolean isEviction() {
    return op.isEviction();
  }

  public void setEvicted() {
    isEvicted = true;
  }

  public boolean isEvicted() {
    return isEvicted;
  }

  public boolean hasRetried() {
    return hasRetried;
  }

  public void setRetried(boolean retried) {
    hasRetried = retried;
  }

  public boolean isPendingSecondaryExpireDestroy() {
    return isPendingSecondaryExpireDestroy;
  }

  public void setPendingSecondaryExpireDestroy(boolean value) {
    isPendingSecondaryExpireDestroy = value;
  }

  // Note that isOriginRemote is sometimes set to false even though the event
  // was received from a peer. This is done to force distribution of the
  // message to peers and to cause concurrency version stamping to be performed.
  // This is done by all one-hop operations, like RemoteInvalidateMessage.
  @Override
  public boolean isOriginRemote() {
    return testEventFlag(EventFlags.FLAG_ORIGIN_REMOTE);
  }

  /* return whether this event originated from a WAN gateway and carries a WAN version tag */
  public boolean isFromWANAndVersioned() {
    return (versionTag != null && versionTag.isGatewayTag());
  }

  /* return whether this event originated in a client and carries a version tag */
  public boolean isFromBridgeAndVersioned() {
    return (context != null) && (versionTag != null);
  }

  @Override
  public boolean isGenerateCallbacks() {
    return testEventFlag(EventFlags.FLAG_GENERATE_CALLBACKS);
  }

  public void setNewEventId(DistributedSystem sys) {
    Assert.assertTrue(eventID == null, "Double setting event id");
    EventID newID = new EventID(sys);
    if (eventID != null) {
      if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
        logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "Replacing event ID with {} in event {}",
            newID, this);
      }
    }
    eventID = newID;
  }

  public void reserveNewEventId(DistributedSystem sys, int count) {
    Assert.assertTrue(eventID == null, "Double setting event id");
    eventID = new EventID(sys);
    if (count > 1) {
      eventID.reserveSequenceId(count - 1);
    }
  }

  public void setEventId(EventID id) {
    eventID = id;
  }

  /**
   * Return the event id, if any
   *
   * @return null if no event id has been set
   */
  @Override
  public EventID getEventId() {
    return eventID;
  }

  @Override
  public boolean isBridgeEvent() {
    return hasClientOrigin();
  }

  @Override
  public boolean hasClientOrigin() {
    return getContext() != null;
  }

  /**
   * sets the ID of the client that initiated this event
   */
  public void setContext(ClientProxyMembershipID contx) {
    Assert.assertTrue(contx != null);
    context = contx;
  }

  /**
   * gets the ID of the client that initiated this event. Null if a server-initiated event
   */
  @Override
  public ClientProxyMembershipID getContext() {
    return context;
  }

  public boolean isLocalInvalid() {
    return testEventFlag(EventFlags.FLAG_LOCAL_INVALID);
  }

  /////////////////////////////////////////////////

  /**
   * Returns the key.
   *
   * @return the key.
   */
  @Override
  public Object getKey() {
    return keyInfo.getKey();
  }

  /**
   * Returns the value in the cache prior to this event. When passed to an event handler after an
   * event occurs, this value reflects the value that was in the cache in this VM, not necessarily
   * the value that was in the cache VM that initiated the operation.
   *
   * @return the value in the cache prior to this event.
   */
  @Override
  public Object getOldValue() {
    try {
      if (isOriginRemote() && getRegion().isProxy()) {
        return null;
      }
      @Unretained
      Object ov = handleNotAvailableOldValue();
      if (ov != null) {
        boolean doCopyOnRead = getRegion().isCopyOnRead();
        if (ov instanceof CachedDeserializable) {
          return callWithOffHeapLock((CachedDeserializable) ov, oldValueCD -> {
            if (doCopyOnRead) {
              return oldValueCD.getDeserializedWritableCopy(getRegion(), re);
            } else {
              return oldValueCD.getDeserializedValue(getRegion(), re);
            }
          });
        } else {
          if (doCopyOnRead) {
            return CopyHelper.copy(ov);
          } else {
            return ov;
          }
        }
      }
      return null;
    } catch (IllegalArgumentException i) {
      IllegalArgumentException iae = new IllegalArgumentException(String.format("%s",
          "Error while deserializing value for key=" + getKey()), i);
      throw iae;
    }
  }

  /**
   * returns the old value after handling one this is NOT_AVAILABLE. If the old value is
   * NOT_AVAILABLE then it may try to read it from disk. If it can't read an unavailable old value
   * from disk then it will return null instead of NOT_AVAILABLE.
   */
  @Unretained(ENTRY_EVENT_OLD_VALUE)
  private Object handleNotAvailableOldValue() {
    @Unretained
    Object result = basicGetOldValue();
    if (result != Token.NOT_AVAILABLE) {
      return result;
    }
    if (getReadOldValueFromDisk()) {
      try {
        result = getRegion().getValueInVMOrDiskWithoutFaultIn(getKey());
      } catch (EntryNotFoundException ex) {
        result = null;
      }
    }
    result = AbstractRegion.handleNotAvailable(result);
    return result;
  }

  /**
   * If true then when getOldValue is called if the NOT_AVAILABLE is found then an attempt will be
   * made to read the old value from disk without faulting it in. Should only be set to true when
   * product is calling a method on a CacheWriter.
   */
  private boolean readOldValueFromDisk;

  public boolean getReadOldValueFromDisk() {
    return readOldValueFromDisk;
  }

  public void setReadOldValueFromDisk(boolean v) {
    readOldValueFromDisk = v;
  }

  /**
   * Like getRawNewValue except that if the result is an off-heap reference then copy it to the
   * heap. Note: to prevent the heap copy use getRawNewValue instead
   */
  public Object getRawNewValueAsHeapObject() {
    Object result = basicGetNewValue();
    if (mayHaveOffHeapReferences()) {
      result = OffHeapHelper.copyIfNeeded(result, getRegion().getCache());
    }
    return result;
  }

  /**
   * If new value is off-heap return the StoredObject form (unretained OFF_HEAP_REFERENCE). Its
   * refcount is not inced by this call and the returned object can only be safely used for the
   * lifetime of the EntryEventImpl instance that returned the value. Else return the raw form.
   */
  @Unretained(ENTRY_EVENT_NEW_VALUE)
  public Object getRawNewValue() {
    return basicGetNewValue();
  }

  @Unretained(ENTRY_EVENT_NEW_VALUE)
  public Object getValue() {
    return basicGetNewValue();
  }

  @Released(ENTRY_EVENT_NEW_VALUE)
  protected void basicSetNewValue(@Retained(ENTRY_EVENT_NEW_VALUE) Object v,
      boolean clearCachedSerializedAndBytes) {
    if (v == newValue) {
      return;
    }
    if (mayHaveOffHeapReferences()) {
      if (offHeapOk) {
        OffHeapHelper.releaseAndTrackOwner(newValue, this);
      }
      if (StoredObject.isOffHeapReference(v)) {
        ReferenceCountHelper.setReferenceCountOwner(this);
        if (!((StoredObject) v).retain()) {
          ReferenceCountHelper.setReferenceCountOwner(null);
          newValue = null;
          return;
        }
        ReferenceCountHelper.setReferenceCountOwner(null);
      }
    }
    newValue = v;
    if (clearCachedSerializedAndBytes) {
      newValueBytes = null;
      cachedSerializedNewValue = null;
    }
  }

  private void generateNewValueFromBytesIfNeeded() {
    if (newValue != null) {
      // no need to generate a new value
      return;
    }
    byte[] bytes = newValueBytes;
    if (bytes != null) {
      newValue = CachedDeserializableFactory.create(bytes, getRegion().getCache());
    }
  }

  @Override
  @Unretained
  public Object basicGetNewValue() {
    generateNewValueFromBytesIfNeeded();
    Object result = newValue;
    if (!offHeapOk && isOffHeapReference(result)) {
      // this.region.getCache().getLogger().info("DEBUG new value already freed " +
      // System.identityHashCode(result));
      throw new IllegalStateException(
          "Attempt to access off heap value after the EntryEvent was released.");
    }
    return result;
  }

  private boolean isOffHeapReference(Object ref) {
    return mayHaveOffHeapReferences() && StoredObject.isOffHeapReference(ref);
  }

  private class OldValueOwner {
    private EntryEventImpl getEvent() {
      return EntryEventImpl.this;
    }

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

    @Override
    public boolean equals(Object obj) {
      if (obj instanceof OldValueOwner) {
        return getEvent().equals(((OldValueOwner) obj).getEvent());
      } else {
        return false;
      }
    }

    @Override
    public String toString() {
      return "OldValueOwner " + getEvent().toString();
    }
  }

  /**
   * Note if v might be an off-heap reference that you did not retain for this EntryEventImpl then
   * call retainsAndSetOldValue instead of this method.
   *
   * @param v the caller should have already retained this off-heap reference.
   */
  @Released(ENTRY_EVENT_OLD_VALUE)
  void basicSetOldValue(@Unretained(ENTRY_EVENT_OLD_VALUE) Object v) {
    @Released
    final Object curOldValue = oldValue;
    if (v == curOldValue) {
      return;
    }
    if (offHeapOk && mayHaveOffHeapReferences()) {
      if (ReferenceCountHelper.trackReferenceCounts()) {
