@Override
  protected boolean operateOnRegion(ClusterDistributionManager dm, LocalRegion r, long startTime)
      throws EntryExistsException, RemoteOperationException {
    setInternalDs(r.getSystem());// set the internal DS. Required to
                                 // checked DS level delta-enabled property
                                 // while sending delta
    boolean sendReply = true;

    InternalDistributedMember eventSender = originalSender;
    if (eventSender == null) {
      eventSender = getSender();
    }
    @Released
    EntryEventImpl eei = EntryEventImpl.create(r, getOperation(), getKey(), null, /* newValue */
        getCallbackArg(),
        useOriginRemote, /* originRemote - false to force distribution in buckets */
        eventSender, true/* generateCallbacks */, false/* initializeId */);
    event = eei;
    try {
      if (versionTag != null) {
        versionTag.replaceNullIDs(getSender());
        event.setVersionTag(versionTag);
      }
      event.setCausedByMessage(this);

      event.setPossibleDuplicate(possibleDuplicate);
      if (bridgeContext != null) {
        event.setContext(bridgeContext);
      }

      Assert.assertTrue(eventId != null);
      event.setEventId(eventId);

      // added for cq procesing
      if (hasOldValue) {
        if (oldValueIsSerialized) {
          event.setSerializedOldValue(getOldValueBytes());
        } else {
          event.setOldValue(getOldValueBytes());
        }
      }

      if (applyDeltaBytes) {
        event.setNewValue(valObj);
        event.setDeltaBytes(deltaBytes);
      } else {
        switch (deserializationPolicy) {
          case DistributedCacheOperation.DESERIALIZATION_POLICY_LAZY:
            event.setSerializedNewValue(getValBytes());
            break;
          case DistributedCacheOperation.DESERIALIZATION_POLICY_NONE:
            event.setNewValue(getValBytes());
            break;
          default:
            throw new AssertionError("unknown deserialization policy: " + deserializationPolicy);
        }
      }

      try {
        result = r.getDataView().putEntry(event, ifNew, ifOld, expectedOldValue,
            requireOldValue, lastModified, true);

        if (!result) { // make sure the region hasn't gone away
          r.checkReadiness();
          if (!ifNew && !ifOld) {
            // no reason to be throwing an exception, so let's retry
            RemoteOperationException ex = new RemoteOperationException(
                "unable to perform put, but operation should not fail");
            sendReply(getSender(), getProcessorId(), dm, new ReplyException(ex), r, startTime);
            return false;
          }
        }
      } catch (CacheWriterException cwe) {
        sendReply(getSender(), getProcessorId(), dm, new ReplyException(cwe), r, startTime);
        return false;
      }

      setOperation(event.getOperation()); // set operation for reply message

      if (sendReply) {
        sendReply(getSender(), getProcessorId(), dm, null, r, startTime, event);
      }
      return false;
    } finally {
      event.release(); // OFFHEAP this may be too soon to make this call
    }
  }
This method is called upon receipt and make the desired changes to the Replicate Region. Note:
It is very important that this message does NOT cause any deadlocks as the sender will wait
indefinitely for the acknowledgement
geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemotePutMessage.java