protected boolean chunkEntries(DistributedRegion rgn, int chunkSizeInBytes,
        boolean includeValues, RegionVersionVector versionVector, Set<Object> unfinishedKeys,
        InitialImageFlowControl flowControl, ObjectIntProcedure proc) throws IOException {
      int MAX_ENTRIES_PER_CHUNK = chunkSizeInBytes / 100;
      if (MAX_ENTRIES_PER_CHUNK < 1000) {
        MAX_ENTRIES_PER_CHUNK = 1000;
      }


      final List<Entry> chunkEntries =
          new InitialImageVersionedEntryList(rgn.getConcurrencyChecksEnabled(),
              MAX_ENTRIES_PER_CHUNK);

      final DiskRegion dr = rgn.getDiskRegion();
      final ByteArrayDataInput in;
      if (dr != null) {
        dr.setClearCountReference();
        in = new ByteArrayDataInput();
      } else {
        in = null;
      }

      VersionSource myId = rgn.getVersionMember();
      Set<VersionSource> foundIds = new HashSet<>();

      if (internalDuringPackingImage != null
          && regionPath.endsWith(internalDuringPackingImage.getRegionName())) {
        internalDuringPackingImage.run();
      }

      try {
        final Iterator<?> it;
        if (versionVector != null) {
          // deltaGII
          it = rgn.entries.regionEntries().iterator();
        } else {
          it = rgn.getBestIterator(includeValues);
        }

        final KnownVersion knownVersion = Versioning
            .getKnownVersionOrDefault(sender.getVersion(), KnownVersion.CURRENT);

        boolean keepGoing;
        boolean sentLastChunk;
        do {
          flowControl.acquirePermit();
          int currentChunkSize = 0;

          while (chunkEntries.size() < MAX_ENTRIES_PER_CHUNK && currentChunkSize < chunkSizeInBytes
              && it.hasNext()) {
            RegionEntry mapEntry = (RegionEntry) it.next();
            Object key = mapEntry.getKey();
            if (rgn.checkEntryNotValid(mapEntry)) { // entry was just removed
              continue;
            }
            if (logger.isDebugEnabled()) {
              Object v = mapEntry.getValueInVM(rgn); // OFFHEAP: noop
              if (v instanceof Conflatable) {
                if (((Conflatable) v).getEventId() == null) {
                  logger.debug("bug 44959: chunkEntries found conflatable with no eventID: {}", v);
                }
              }
            }
            final InitialImageOperation.Entry entry;
            if (includeValues) {
              final boolean fillRes;
              try {
                // also fills in lastModifiedTime
                VersionStamp<?> stamp = mapEntry.getVersionStamp();
                if (stamp != null) {
                  synchronized (mapEntry) { // must sync to make sure the tag goes with the value
                    VersionSource<?> id = stamp.getMemberID();
                    if (id == null) {
                      id = myId;
                    }
                    foundIds.add(id);
                    // if the recipient passed a version vector, use it to filter out
                    // entries the recipient already has
                    // For keys in unfinishedKeys, not to filter them out
                    if ((unfinishedKeys == null || !unfinishedKeys.contains(key))
                        && versionVector != null) {
                      if (versionVector.contains(id, stamp.getRegionVersion())) {
                        continue;
                      }
                    }
                    entry = new InitialImageOperation.Entry();
                    entry.key = key;
                    entry.setVersionTag(stamp.asVersionTag());
                    fillRes = mapEntry.fillInValue(rgn, entry, in, rgn.getDistributionManager(),
                        knownVersion);
                    if (versionVector != null) {
                      if (logger.isTraceEnabled(LogMarker.INITIAL_IMAGE_VERBOSE)) {
                        logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                            "chunkEntries:entry={},stamp={}", entry, stamp);
                      }
                    }
                  }
                } else {
                  entry = new InitialImageOperation.Entry();
                  entry.key = key;
                  fillRes = mapEntry.fillInValue(rgn, entry, in, rgn.getDistributionManager(),
                      knownVersion);
                }
              } catch (DiskAccessException dae) {
                rgn.handleDiskAccessException(dae);
                throw dae;
              }
              if (!fillRes) {
                // map entry went away
                continue;
              }
            } else {
              entry = new InitialImageOperation.Entry();
              entry.key = key;
              entry.setLocalInvalid();
              entry.setLastModified(mapEntry.getLastModified());
            }

            chunkEntries.add(entry);
            currentChunkSize += entry.calcSerializedSize();
          }

          // send 1 for last message if no more data
          int lastMsg = it.hasNext() ? 0 : 1;
          keepGoing = proc.executeWith(chunkEntries, lastMsg);
          sentLastChunk = lastMsg == 1 && keepGoing;
          chunkEntries.clear();

          // if this region is destroyed while we are sending data, then abort.
        } while (keepGoing && it.hasNext());

        RegionVersionVector vv = rgn.getVersionVector();
        if (vv != null) {
          vv.removeOldMembers(foundIds);
        }
        // return false if we were told to abort
        return sentLastChunk;
      } finally {
        if (dr != null) {
          dr.removeClearCountReference();
        }
      }
    }
Serialize the entries into byte[] chunks, calling proc for each one. proc args: the byte[]
chunk and an int indicating whether it is the last chunk (positive means last chunk, zero
otherwise). The return value of proc indicates whether to continue to the next chunk (true)
or abort (false).

@param versionVector requester's region version vector
@param unfinishedKeys keys of unfinished operation (persistent region only)
@return true if finished all chunks, false if stopped early
geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java