void executeOnRegion(DistributedRegionFunctionStreamingMessage msg, final Function function,
      final Object args, int prid, final Set filter, boolean isReExecute) throws IOException {
    final DistributionManager dm = getDistributionManager();
    ResultSender resultSender = new DistributedRegionFunctionResultSender(dm, msg, function);
    final RegionFunctionContextImpl context = new RegionFunctionContextImpl(cache, function.getId(),
        this, args, filter, null, null, resultSender, isReExecute);
    FunctionStats stats = FunctionStatsManager.getFunctionStats(function.getId(), dm.getSystem());
    long start = stats.startFunctionExecution(function.hasResult());
    try {
      function.execute(context);
      stats.endFunctionExecution(start, function.hasResult());
    } catch (FunctionException functionException) {
      if (logger.isDebugEnabled()) {
        logger.debug("FunctionException occurred on remote node  while executing Function: {}",
            function.getId(), functionException);
      }
      stats.endFunctionExecutionWithException(start, function.hasResult());
      throw functionException;
    } catch (CacheClosedException cacheClosedexception) {
      if (logger.isDebugEnabled()) {
        logger.debug("CacheClosedException occurred on remote node  while executing Function: {}",
            function.getId(), cacheClosedexception);
      }
      throw cacheClosedexception;
    } catch (Exception exception) {
      if (logger.isDebugEnabled()) {
        logger.debug("Exception occurred on remote node  while executing Function: {}",
            function.getId(), exception);
      }
      stats.endFunctionExecutionWithException(start, function.hasResult());
      throw new FunctionException(exception);
    }
  }
@return a random persistent replicate, null if there is none
geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java