@Override
    protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm,
        PartitionedRegion pr, long startTime)
        throws CacheException, QueryException, ForceReattemptException, InterruptedException {
      final boolean isTraceEnabled = logger.isTraceEnabled();

      if (Thread.interrupted()) {
        throw new InterruptedException();
      }
      Object nextObject;
      Object failedObject = null;
      int socketBufferSize = dm.getSystem().getConfig().getSocketBufferSize();
      int chunkSize = socketBufferSize - StreamingOperation.MSG_OVERHEAD;
      boolean sentFinalMessage = false;
      boolean receiverCacheClosed = false;

      outStream = new HeapDataOutputStream(chunkSize, Versioning
          .getKnownVersionOrDefault(getSender().getVersion(), KnownVersion.CURRENT));

      try {
        do {
          // boolean firstObject = true;

          // always write at least one object, allowing expansion
          // if we have an object already that didn't get added, then use
          // that object instead of getting another one
          if (failedObject == null) {
            nextObject = getNextReplyObject(pr);
            replyLastMsg = nextObject == Token.END_OF_STREAM;
          } else {
            nextObject = failedObject;
            failedObject = null;
          }

          if (!replyLastMsg) {
            numObjectsInChunk = 1;
            if (isTraceEnabled) {
              logger.trace("Writing this object to StreamingPartitionMessage outStream: '{}'",
                  nextObject);
            }
            BlobHelper.serializeTo(nextObject, outStream);

            // for the next objects, disallow stream from allocating more storage
            do {
              outStream.disallowExpansion(CHUNK_FULL); // sets the mark where rollback occurs on
                                                       // CHUNK_FULL

              nextObject = getNextReplyObject(pr);
              replyLastMsg = nextObject == Token.END_OF_STREAM;

              if (!replyLastMsg) {
                try {

                  if (isTraceEnabled) {
                    logger.trace("Writing this object to StreamingPartitionMessage outStream: '{}'",
                        nextObject);
                  }
                  BlobHelper.serializeTo(nextObject, outStream);
                  numObjectsInChunk++;
                } catch (GemFireRethrowable e) {
                  // can only be thrown when expansion is disallowed
                  // and buffer is automatically reset to point where it was disallowed
                  failedObject = nextObject;
                  break;
                }
              }
            } while (nextObject != Token.END_OF_STREAM);
          }

          try {
            sendReply(getSender(), processorId, dm, null, pr, startTime);
            replyMsgNum++;
            if (replyLastMsg) {
              sentFinalMessage = true;
            }
          } catch (CancelException e) {
            receiverCacheClosed = true;
            break;
          }
          outStream.reset(); // ready for reuse, assumes sendReply
                             // does not queue the message but outStream has
                             // already been used
          numObjectsInChunk = 0;
        } while (!replyLastMsg);
      } catch (IOException ioe) {
        // not expected to ever happen
        throw new InternalGemFireException(ioe);
      }


      if (!sentFinalMessage && !receiverCacheClosed) {
        throw new InternalGemFireError(
            "unexpected condition");
      }

      // otherwise, we're already done, so don't send another reply
      return false;
    }
An operation upon the messages partitioned region

@param dm the manager that received the message
@param pr the partitioned region that should be modified
@return true if a reply message should be sent
@throws CacheException if an error is generated in the remote cache
@throws ForceReattemptException if the peer is no longer available
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/StreamingPartitionOperation.java