@Override
  public void process(DistributionMessage msg) {
    if (!waitingOnMember(msg.getSender())) {
      return;
    }
    msgsBeingProcessed.incrementAndGet();
    try {
      ReplyMessage m = (ReplyMessage) msg;
      if (m.getException() == null) {
        FunctionStreamingReplyMessage functionReplyMsg = (FunctionStreamingReplyMessage) m;
        Object result = functionReplyMsg.getResult();
        boolean isLast = false;
        synchronized (processSingleResult) {
          isLast = trackMessage(functionReplyMsg);
          functionResultWaiter.processData(result, isLast, msg.getSender());
        }
        if (isLast) {
          super.process(msg, false); // removes from members and cause us
          // to ignore future messages received from that member
        }
      } else {
        if (execution.forwardExceptions || (execution.waitOnException
        /* && !(m.getException().getCause() instanceof BucketMovedException) */)) {
          // send BucketMovedException forward which will be handled by LocalResultCollectorImpl
          synchronized (processSingleResult) {
            functionResultWaiter.processData(m.getException().getCause(), true,
                msg.getSender());
          }
        }
        super.process(msg, false);
      }
    } finally {
      msgsBeingProcessed.decrementAndGet();
      checkIfDone(); // check to see if decrementing msgsBeingProcessed requires signalling to
                     // proceed
    }
  }
Return true if this is the very last reply msg to process for this member
geode-core/src/main/java/org/apache/geode/internal/cache/execute/FunctionStreamingResultCollector.java