public static void sendReply(PartitionedRegion pr, final InternalDistributedMember recipient,
        final int processorId, final DistributionManager dm,
        final HashMap<Integer, HashSet> bucketKeys, final HashSet<Integer> bucketIds, String regex,
        boolean allowTombstones, long startTime) throws ForceReattemptException {

      PartitionedRegionDataStore ds = pr.getDataStore();
      if (ds == null) {
        return;
      }
      ArrayList<BucketRegion> maps = new ArrayList<>();
      HashSet<Integer> failedBuckets = new HashSet<>();

      Set<Integer> bucketIdSet = null;
      if (bucketKeys != null) {
        bucketIdSet = bucketKeys.keySet();
      } else { // bucketIds != null
        bucketIdSet = bucketIds;
      }
      for (int id : bucketIdSet) {
        try {
          maps.add(ds.handleRemoteGetEntries(id));
        } catch (ForceReattemptException fre) {
          failedBuckets.add(id);
        }
      }

      BucketRegion map = null;
      boolean lockAcquired = false;
      try (HeapDataOutputStream mos =
          new HeapDataOutputStream(InitialImageOperation.CHUNK_SIZE_IN_BYTES + 2048,
              Versioning.getKnownVersionOrDefault(recipient.getVersion(), KnownVersion.CURRENT))) {
        Iterator<BucketRegion> mapsIterator = maps.iterator();
        Iterator it = null;

        boolean keepGoing = true;
        boolean writeFooter = false;
        boolean lastMsgSent = false;
        boolean needToWriteBucketInfo = true;
        int msgNum = 0;

        while (mapsIterator.hasNext()) {
          if (map != null && lockAcquired) {
            try {
              map.releaseDestroyLock();
              // instead take a bucketCreationLock.getWriteLock() or pr.BucketLock?
            } catch (CancelException ignored1) {
            } finally {
              lockAcquired = false;
            }
          }
          map = mapsIterator.next();
          if (map.isBucketDestroyed()) {
            failedBuckets.add(map.getId());
            continue;
          }
          try {
            map.acquireDestroyLock();
            lockAcquired = true;
          } catch (CancelException e) {
            if (logger.isDebugEnabled()) {
              logger.debug("sendReply: acquireDestroyLock failed due to cache closure, region = {}",
                  map.getFullPath());
            }
          }

          try {
            if (bucketKeys != null) {
              it = bucketKeys.get(map.getId()).iterator();
            } else { // bucketIds != null
              if (regex == null) {
                it = new HashSet(map.keySet(allowTombstones)).iterator();
              } else {
                it =
                    map.getKeysWithInterest(InterestType.REGULAR_EXPRESSION, regex, allowTombstones)
                        .iterator();
              }
            }

            while (it.hasNext()) {
              Object key = it.next();
              VersionTagHolder clientEvent = new VersionTagHolder();
              Object value =
                  map.get(key, null, true, true, true, null, clientEvent, allowTombstones);

              if (needToWriteBucketInfo) {
                DataSerializer.writePrimitiveInt(map.getId(), mos);
                needToWriteBucketInfo = false;
                writeFooter = true;
              }

              int entrySize = mos.size();
              DataSerializer.writeObject(key, mos);
              VersionTag versionTag = clientEvent.getVersionTag();
              if (versionTag != null) {
                versionTag.replaceNullIDs(map.getVersionMember());
              }
              DataSerializer.writeObject(value, mos);
              DataSerializer.writeObject(versionTag, mos);
              entrySize = mos.size() - entrySize;

              // If no more space OR no more entries in bucket, write end-of-bucket marker.
              if ((mos.size() + entrySize) >= InitialImageOperation.CHUNK_SIZE_IN_BYTES
                  || !it.hasNext()) {

                DataSerializer.writeObject(null, mos);
                DataSerializer.writePrimitiveBoolean(it.hasNext(), mos);
                needToWriteBucketInfo = true;
                writeFooter = false; // Be safe
              }

              // If no more space in chunk, send it.
              if ((mos.size() + entrySize) >= InitialImageOperation.CHUNK_SIZE_IN_BYTES) {

                // Send as last message if no more data
                boolean lastMsg = !(it.hasNext() || mapsIterator.hasNext());

                ++msgNum;
                FetchBulkEntriesReplyMessage reply =
                    new FetchBulkEntriesReplyMessage(recipient, processorId, mos, msgNum, lastMsg);
                if (lastMsg) {
                  reply.failedBucketIds = failedBuckets;
                }
                Set failures = dm.putOutgoing(reply);
                keepGoing = (failures == null) || (failures.size() == 0);
                if (lastMsg && keepGoing) {
                  lastMsgSent = true;
                }
                mos.reset();

              } // else still enough space
            } // while (for each key)

            if (!keepGoing) {
              throw new ForceReattemptException("Failed to send response");
            }
          } catch (IOException ioe) {
            throw new ForceReattemptException(
                "Unable to send response to fetch-entries request",
                ioe);
          } finally {
            if (lockAcquired) {
              try {
                map.releaseDestroyLock();
              } catch (CancelException ignored1) {
              } finally {
                lockAcquired = false;
              }
            }
          }
        } // while (for each map)

        if (!lastMsgSent) {
          if (mos.size() == 0) {
            try {
              DataSerializer.writePrimitiveInt(-1, mos);
            } catch (IOException ioe) {
              throw new ForceReattemptException(
                  "Unable to send response to fetch-entries request",
                  ioe);
            }
          } else if (writeFooter) {
            try {
              DataSerializer.writeObject(null, mos); // end of entries of current bucket in current
              // response
              DataSerializer.writePrimitiveBoolean(false, mos); // no more entries of current bucket
            } catch (IOException ioe) {
              throw new ForceReattemptException(
                  "Unable to send response to fetch-entries request",
                  ioe);
            }
          }
          ++msgNum;
          FetchBulkEntriesReplyMessage reply =
              new FetchBulkEntriesReplyMessage(recipient, processorId, mos, msgNum, true);
          reply.failedBucketIds = failedBuckets;
          Set failures = dm.putOutgoing(reply);
          if (failures != null && failures.size() > 0) {
            throw new ForceReattemptException("Failed to send response");
          }
        }
      }

      if (lockAcquired) {
        try {
          map.releaseDestroyLock();
        } catch (CancelException e) {
          // ignore
        } finally {
          lockAcquired = false;
        }
      }
    }
Empty constructor to conform to DataSerializable interface
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/FetchBulkEntriesMessage.java