@Override
  protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion r,
      long startTime) throws EntryExistsException, DataLocationException, IOException {
    setInternalDs(r.getSystem());// set the internal DS. Required to
                                 // checked DS level delta-enabled property
                                 // while sending delta
    PartitionedRegionDataStore ds = r.getDataStore();
    boolean sendReply = true;

    InternalDistributedMember eventSender = originalSender;
    if (eventSender == null) {
      eventSender = getSender();
    }
    @Released
    final EntryEventImpl ev =
        EntryEventImpl.create(r, getOperation(), getKey(), null, /* newValue */
            getCallbackArg(), false/* originRemote - false to force distribution in buckets */,
            eventSender, generateCallbacks/* generateCallbacks */, false/* initializeId */);
    try {
      if (versionTag != null) {
        versionTag.replaceNullIDs(getSender());
        ev.setVersionTag(versionTag);
      }
      if (bridgeContext != null) {
        ev.setContext(bridgeContext);
      }
      Assert.assertTrue(eventId != null);
      ev.setEventId(eventId);
      ev.setCausedByMessage(this);
      ev.setInvokePRCallbacks(!notificationOnly);
      ev.setPossibleDuplicate(posDup);
      /*
       * if (this.hasOldValue) { if (this.oldValueIsSerialized) {
       * ev.setSerializedOldValue(getOldValueBytes()); } else { ev.setOldValue(getOldValueBytes());
       * } }
       */

      ev.setDeltaBytes(deltaBytes);
      if (hasDelta) {
        valObj = null;
        // New value will be set once it is generated with fromDelta() inside
        // EntryEventImpl.processDeltaBytes()
        ev.setNewValue(valObj);
      } else {
        switch (deserializationPolicy) {
          case DistributedCacheOperation.DESERIALIZATION_POLICY_LAZY:
            ev.setSerializedNewValue(getValBytes());
            break;
          case DistributedCacheOperation.DESERIALIZATION_POLICY_NONE:
            ev.setNewValue(getValBytes());
            break;
          default:
            throw new AssertionError("unknown deserialization policy: " + deserializationPolicy);
        }
      }

      if (!notificationOnly) {
        if (ds == null) {
          throw new AssertionError(
              "This process should have storage" + " for this operation: " + this);
        }
        try {
          ev.setOriginRemote(false);
          result =
              r.getDataView().putEntryOnRemote(ev, ifNew, ifOld, expectedOldValue,
                  requireOldValue, lastModified, true/* overwriteDestroyed *not* used */);

          if (!result) { // make sure the region hasn't gone away
            r.checkReadiness();
          }
        } catch (CacheWriterException | PrimaryBucketException cwe) {
          sendReply(getSender(), getProcessorId(), dm, new ReplyException(cwe), r, startTime);
          return false;
        } catch (InvalidDeltaException ide) {
          sendReply(getSender(), getProcessorId(), dm, new ReplyException(ide), r, startTime);
          r.getCachePerfStats().incDeltaFullValuesRequested();
          return false;
        }
        if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
          logger.trace(LogMarker.DM_VERBOSE, "PutMessage {} with key: {} val: {}",
              (result ? "updated bucket" : "did not update bucket"), getKey(),
              (getValBytes() == null ? "null" : "(" + getValBytes().length + " bytes)"));
        }
      } else { // notificationOnly
        @Released
        EntryEventImpl e2 = createListenerEvent(ev, r, dm.getDistributionManagerId());
        final EnumListenerEvent le;
        try {
          if (e2.getOperation().isCreate()) {
            le = EnumListenerEvent.AFTER_CREATE;
          } else {
            le = EnumListenerEvent.AFTER_UPDATE;
          }
          r.invokePutCallbacks(le, e2, r.isInitialized(), true);
        } finally {
          // if e2 == ev then no need to free it here. The outer finally block will get it.
          if (e2 != ev) {
            e2.release();
          }
        }
        result = true;
      }

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

      if (sendReply) {
        sendReply(getSender(), getProcessorId(), dm, null, r, startTime, ev);
      }
      return false;
    } finally {
      ev.release();
    }
  }
This method is called upon receipt and make the desired changes to the PartitionedRegion 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/partitioned/PutMessage.java