@Override
    protected void process(ClusterDistributionManager dm) {
      long startTime = dm.cacheTimeMillis();
      final InitializationLevel oldLevel =
          LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);
      try {
        InternalCache gfc = dm.getExistingCache();
        LocalRegion region = (LocalRegion) gfc.getRegion(regionName);
        if (region != null && region.isInitialized()
            && (dm.cacheTimeMillis() - startTime < timeoutMs)) {
          CacheWriter writer = region.basicGetWriter();
          EntryEventImpl entryEvtImpl = null;
          RegionEventImpl regionEvtImpl = null;
          if (event instanceof EntryEventImpl) {
            entryEvtImpl = (EntryEventImpl) event;
            entryEvtImpl.setRegion(region);
            Operation op = entryEvtImpl.getOperation();
            if (op == Operation.REPLACE) {
              entryEvtImpl.setOperation(Operation.UPDATE);
            } else if (op == Operation.PUT_IF_ABSENT) {
              entryEvtImpl.setOperation(Operation.CREATE);
            } else if (op == Operation.REMOVE) {
              entryEvtImpl.setOperation(Operation.DESTROY);
            }
            // [bruce] even if this event was transmitted from another VM, its operation may have
            // originated in this VM. PartitionedRegion.put() with a cacheWriter is one
            // situation where this can occur
            entryEvtImpl.setOriginRemote(event.getDistributedMember() == null
                || !event.getDistributedMember().equals(dm.getDistributionManagerId()));
          } else if (event instanceof RegionEventImpl) {
            regionEvtImpl = (RegionEventImpl) event;
            regionEvtImpl.region = region;
            regionEvtImpl.originRemote = true;
          }

          if (writer != null) {
            if (entryEvtImpl != null) {
              entryEvtImpl.setReadOldValueFromDisk(true);
            }
            try {
              switch (action) {
                case BEFORECREATE:
                  writer.beforeCreate(entryEvtImpl);
                  break;
                case BEFOREDESTROY:
                  writer.beforeDestroy(entryEvtImpl);
                  break;
                case BEFOREUPDATE:
                  writer.beforeUpdate(entryEvtImpl);
                  break;
                case BEFOREREGIONDESTROY:
                  writer.beforeRegionDestroy(regionEvtImpl);
                  break;
                case BEFOREREGIONCLEAR:
                  writer.beforeRegionClear(regionEvtImpl);
                  break;
                default:
                  break;

              }
              NetWriteReplyMessage.sendMessage(getSender(), processorId,
                  dm, true, null, false);
            } catch (CacheWriterException cwe) {
              NetWriteReplyMessage.sendMessage(getSender(), processorId,
                  dm, false, cwe, true);
            } catch (Exception e) {
              NetWriteReplyMessage.sendMessage(getSender(), processorId,
                  dm, false, e, false);
            } finally {
              if (entryEvtImpl != null) {
                entryEvtImpl.setReadOldValueFromDisk(false);
              }
            }

          } else {
            NetWriteReplyMessage.sendMessage(getSender(), processorId,
                dm, false,
                new TryAgainException(
                    "No cachewriter defined"),
                true);
          }

        } else {
          NetWriteReplyMessage.sendMessage(getSender(), processorId, dm,
              false,
              new TryAgainException("Timeout expired or region not ready"),
              true);

        }
      } catch (RegionDestroyedException ignore) {
        NetWriteReplyMessage.sendMessage(getSender(), processorId, dm,
            false, null, false);

      } catch (DistributedSystemDisconnectedException e) {
        // shutdown condition
        throw e;
      } catch (CancelException cce) {
        dm.getCancelCriterion().checkCancelInProgress(cce); // TODO anyway to find the region or
                                                            // cache here?
        NetWriteReplyMessage.sendMessage(getSender(), processorId, dm,
            false, null, false);
      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable t) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        NetWriteReplyMessage.sendMessage(getSender(), processorId, dm,
            false,
            new InternalGemFireException(
                "Error processing request",
                t),
            true);
      } finally {
        LocalRegion.setThreadInitLevelRequirement(oldLevel);
      }
    }
Invoked on the node that has the object
geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java