public void executeOnDataStore(final Set localKeys, final Function function, final Object object,
      final int prid, final int[] bucketArray, final boolean isReExecute,
      final PartitionedRegionFunctionStreamingMessage msg, long time, ServerConnection servConn,
      int transactionID, Object principal) {

    if (!areAllBucketsHosted(bucketArray)) {
      throw new BucketMovedException(
          "Bucket migrated to another node. Please retry.");
    }
    final DistributionManager dm = partitionedRegion.getDistributionManager();

    ResultSender resultSender = new PartitionedRegionFunctionResultSender(dm,
        partitionedRegion, time, msg, function, bucketArray);

    final RegionFunctionContextImpl prContext =
        new RegionFunctionContextImpl(getPartitionedRegion().getCache(), function.getId(),
            partitionedRegion, object, localKeys, ColocationHelper
                .constructAndGetAllColocatedLocalDataSet(partitionedRegion, bucketArray),
            bucketArray, resultSender, isReExecute, principal);

    FunctionStats stats = FunctionStatsManager.getFunctionStats(function.getId(), dm.getSystem());
    long start = stats.startFunctionExecution(function.hasResult());
    try {
      if (logger.isDebugEnabled()) {
        logger.debug("Executing Function:{} on Remote Node with context:{}", function.getId(),
            prContext);
      }
      function.execute(prContext);
      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());
      if (functionException.getCause() instanceof QueryInvalidException) {
        // Handle this exception differently since it can contain
        // non-serializable objects.
        // java.io.NotSerializableException: antlr.CommonToken
        // create a new FunctionException on the original one's message (not cause).
        throw new FunctionException(functionException.getLocalizedMessage());
      }
      throw functionException;
    }
  }
Checks consistency of bucket and meta data before attempting to grab the bucket.

@return false if bucket should not be grabbed, else true. TODO prpersist - move this to
        BucketRegion
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java