@Override
  public Object getResultInternal(long timeout, TimeUnit unit)
      throws FunctionException, InterruptedException {
    long timeoutInMillis = unit.toMillis(timeout);
    if (resultCollected) {
      throw new FunctionException("Result already collected");
    }
    resultCollected = true;
    if (hasResult) {
      try {
        long timeBefore = System.currentTimeMillis();
        if (!waitForCacheOrFunctionException(timeoutInMillis)) {
          throw new FunctionException("All results not received in time provided.");
        }
        long timeAfter = System.currentTimeMillis();
        timeoutInMillis = timeoutInMillis - (timeAfter - timeBefore);
        if (timeoutInMillis < 0) {
          timeoutInMillis = 0;
        }

        if (!execution.getFailedNodes().isEmpty() && !execution.isClientServerMode()) {
          // end the rc and clear it
          endResults();
          clearResults();
          execution = execution.setIsReExecute();
          ResultCollector newRc = null;
          if (execution.isFnSerializationReqd()) {
            newRc = execution.execute(fn);
          } else {
            newRc = execution.execute(fn.getId());
          }
          return newRc.getResult(timeoutInMillis, unit);
        }
        if (!execution.getWaitOnExceptionFlag() && fites.size() > 0) {
          throw new FunctionException(fites.get(0));
        }
      } catch (FunctionInvocationTargetException fite) {
        if (!fn.isHA()) {
          throw new FunctionException(fite);
        } else if (execution.isClientServerMode()) {
          clearResults();
          FunctionInvocationTargetException fe = new InternalFunctionInvocationTargetException(
              fite.getMessage(), execution.getFailedNodes());
          throw new FunctionException(fe);
        } else {
          clearResults();
          execution = execution.setIsReExecute();
          ResultCollector newRc = null;
          if (execution.isFnSerializationReqd()) {
            newRc = execution.execute(fn);
          } else {
            newRc = execution.execute(fn.getId());
          }
          return newRc.getResult(timeoutInMillis, unit);
        }
      } catch (BucketMovedException e) {
        if (!fn.isHA()) {
          // endResults();
          FunctionInvocationTargetException fite =
              new FunctionInvocationTargetException(e.getMessage());
          throw new FunctionException(fite);
        } else if (execution.isClientServerMode()) {
          // endResults();
          clearResults();
          FunctionInvocationTargetException fite =
              new FunctionInvocationTargetException(e.getMessage());
          throw new FunctionException(fite);
        } else {
          // endResults();
          clearResults();
          execution = execution.setIsReExecute();
          ResultCollector newRc = null;
          if (execution.isFnSerializationReqd()) {
            newRc = execution.execute(fn);
          } else {
            newRc = execution.execute(fn.getId());
          }
          return newRc.getResult(timeoutInMillis, unit);
        }
      } catch (CacheClosedException e) {
        if (!fn.isHA()) {
          // endResults();
          FunctionInvocationTargetException fite =
              new FunctionInvocationTargetException(e.getMessage());
          throw new FunctionException(fite);
        } else if (execution.isClientServerMode()) {
          // endResults();
          clearResults();
          FunctionInvocationTargetException fite = new InternalFunctionInvocationTargetException(
              e.getMessage(), execution.getFailedNodes());
          throw new FunctionException(fite);
        } else {
          // endResults();
          clearResults();
          execution = execution.setIsReExecute();
          ResultCollector newRc = null;
          if (execution.isFnSerializationReqd()) {
            newRc = execution.execute(fn);
          } else {
            newRc = execution.execute(fn.getId());
          }
          return newRc.getResult(timeoutInMillis, unit);
        }
      } catch (CacheException e) {
        // endResults();
        throw new FunctionException(e);
      } catch (ForceReattemptException e) {
        // this is case of WrapperException which enforce the re execution of
        // the function.
        if (!fn.isHA()) {
          throw new FunctionException(e);
        } else if (execution.isClientServerMode()) {
          clearResults();
          FunctionInvocationTargetException iFITE = new InternalFunctionInvocationTargetException(
              e.getMessage(), execution.getFailedNodes());
          throw new FunctionException(iFITE);
        } else {
          clearResults();
          execution = execution.setIsReExecute();
          ResultCollector newRc = null;
          if (execution.isFnSerializationReqd()) {
            newRc = execution.execute(fn);
          } else {
            newRc = execution.execute(fn.getId());
          }
          return newRc.getResult();
        }
      }
    }
    return userRC.getResult(timeoutInMillis, unit); // As we have already waited for timeout
                                                    // earlier we expect results to be ready
  }
Contract of {@link ReplyProcessor21#stillWaiting()} is that it never returns true after
returning false.
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PRFunctionStreamingResultCollector.java