@Override
  public void process(final ClusterDistributionManager dm) {
    InternalCache cache = getCache(dm);
    if (cache == null) {
      String message = getCacheClosedMessage(dm);
      ReplyException replyException = new ReplyException(new CacheClosedException(message));
      sendReply(getSender(), processorId, dm, replyException, null, 0);
      return;
    }

    if (dm.getSystem().threadOwnsResources()) {
      // reply inline if thread owns socket.
      doRemoteOperation(dm, cache);
      return;
    }

    if (isTransactional()) {
      dm.getExecutors().getWaitingThreadPool().execute(() -> doRemoteOperation(dm, cache));
    } else {
      // reply inline for non-transactional case.
      doRemoteOperation(dm, cache);
    }
  }
Upon receipt of the message, both process the message and send an acknowledgement, not
necessarily in that order. Note: Any hang in this message may cause a distributed deadlock for
those threads waiting for an acknowledgement.

@throws RegionDestroyedException if the region does not exist
geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemoteOperationMessage.java