@Override
  protected void process(final ClusterDistributionManager dm) {
    Throwable thr = null;
    boolean sendReply = true;
    DistributedRegion dr = null;
    TXStateProxy tx = null;

    try {
      if (checkCacheClosing(dm) || checkDSClosing(dm)) {
        InternalCache cache = dm.getCache();
        if (cache != null) {
          thr = cache
              .getCacheClosedException(String.format("Remote cache is closed: %s",
                  dm.getId()));
        } else {
          thr = new CacheClosedException(String.format("Remote cache is closed: %s",
              dm.getId()));
        }
        return;
      }
      dr = (DistributedRegion) dm.getCache().getRegion(regionPath);
      if (dr == null) {
        // if the distributed system is disconnecting, don't send a reply saying
        // the partitioned region can't be found (bug 36585)
        thr = new ForceReattemptException(dm.getDistributionManagerId().toString()
            + ": could not find Distributed region " + regionPath);
        return; // reply sent in finally block below
      }
      thr = UNHANDLED_EXCEPTION;
      tx = prepForTransaction(dm);
      sendReply = operateOnDistributedRegion(dm, dr); // need to take care of
                                                      // it...
      thr = null;
    } catch (CancelException se) {
      // bug 37026: this is too noisy...
      // throw new CacheClosedException("remote system shutting down");
      // thr = se; cache is closed, no point trying to send a reply
      thr = null;
      sendReply = false;
      if (logger.isDebugEnabled()) {
        logger.debug("shutdown caught, abandoning message: {}", se.getMessage(), se);
      }
    } catch (RegionDestroyedException rde) {
      // [bruce] RDE does not always mean that the sender's region is also
      // destroyed, so we must send back an exception. If the sender's
      // region is also destroyed, who cares if we send it an exception
      if (dr != null && dr.isClosed()) {
        thr = new ForceReattemptException("Region is destroyed in " + dm.getDistributionManagerId(),
            rde);
      }
    } 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) {
      logger.debug("{} exception occurred while processing message: {}", this, t.getMessage(), 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();
      // log the exception at fine level if there is no reply to the message
      thr = null;
      if (sendReply && processorId != 0) {
        if (!checkDSClosing(dm)) {
          thr = t;
        } else {
          // don't pass arbitrary runtime exceptions and errors back if this
          // cache/vm is closing
          thr = new ForceReattemptException("Distributed system is disconnecting");
        }
      }
      if (processorId == 0) {
        logger.debug("{} exception while processing message: {}", this, t.getMessage(), t);
      } else if (logger.isTraceEnabled(LogMarker.DM_VERBOSE) && (t instanceof RuntimeException)) {
        logger.trace(LogMarker.DM_VERBOSE, "Exception caught while processing message", t);
      }
    } finally {
      cleanupTransaction(tx);
      if (sendReply && processorId != 0) {
        ReplyException rex = null;
        if (thr != null) {
          // don't transmit the exception if this message was to a listener
          // and this listener is shutting down
          boolean excludeException = false;
          if (!functionObject.isHA()) {
            excludeException =
                (thr instanceof CacheClosedException || (thr instanceof ForceReattemptException));
          } else {
            excludeException = thr instanceof ForceReattemptException;
          }
          if (!excludeException) {
            rex = new ReplyException(thr);
          }
        }
        // Send the reply if the operateOnPartitionedRegion returned true
        sendReply(getSender(), processorId, dm, rex, null, 0, true, false);
      }
    }
  }
default exception to ensure a false-positive response is never returned
geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegionFunctionStreamingMessage.java