boolean processChunk(List<Entry> entries, InternalDistributedMember sender)
      throws IOException, ClassNotFoundException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    final boolean isTraceEnabled = logger.isTraceEnabled();

    // one volatile read of test flag
    int slow = slowImageProcessing;
    final CachePerfStats stats = region.getCachePerfStats();
    ImageState imgState = region.getImageState();
    // Asif : Can the image state be null here. Don't think so
    // Assert.assertTrue(imgState != null, "processChunk :ImageState should not have been null ");
    // Asif: Set the Htree Reference in Thread Local before the iteration begins so as
    // to detect a clear operation occurring while the put operation is in progress
    // It is Ok to set it every time the loop is executed, because a clear can happen
    // only once during GII life cycle & so it does not matter if the HTree ref changes after the
    // clear
    // whenever a conflict is detected in DiskRegion it is Ok to abort the operation
    final DiskRegion diskRegion = region.getDiskRegion();
    if (diskRegion != null) {
      diskRegion.setClearCountReference();
    }
    try {
      int entryCount = entries.size();
      Set<Object> keys = null;
      if (entryCount <= 1000 && isDebugEnabled) {
        keys = new HashSet<>();
      }
      List<Entry> entriesToSynchronize = new ArrayList<>();

      for (int i = 0; i < entryCount; i++) {
        // stream is null-terminated
        if (internalDuringApplyDelta != null && !internalDuringApplyDelta.isRunning
            && internalDuringApplyDelta.getRegionName().equals(region.getName())) {
          internalDuringApplyDelta.run();
        }
        if (slow > 0) {
          // make sure we are still slow
          slow = slowImageProcessing;
          if (slow > 0) {
            boolean interrupted = Thread.interrupted();
            try {
              if (isDebugEnabled) {
                logger.debug("processChunk: Sleeping for {} ms for rgn {}", slow,
                    region.getFullPath());
              }
              Thread.sleep(slow);
              slowImageSleeps.getAndIncrement();
            } catch (InterruptedException e) {
              interrupted = true;
              region.getCancelCriterion().checkCancelInProgress(e);
            } finally {
              if (interrupted) {
                Thread.currentThread().interrupt();
              }
            }
          }
        }
        try {
          if (region.isDestroyed() || imgState.getClearRegionFlag()) {
            return false;
          }
        } catch (CancelException e) {
          return false;
        }

        Entry entry = entries.get(i);

        stats.incGetInitialImageKeysReceived();

        final long lastModified = entry.getLastModified();

        Object tmpValue = entry.value;

        boolean didIIP = false;
        boolean wasRecovered = false;

        VersionTag tag = entry.getVersionTag();

        if (diskRegion != null) {
          // verify if entry from GII is the same as the one from recovery
          RegionEntry regionEntry = this.entries.getEntry(entry.key);
          if (isTraceEnabled) {
            logger.trace("processChunk:entry={},tag={},re={}", entry, tag, regionEntry);
          }
          // re will be null if the gii chunk gives us a create
          if (regionEntry != null) {
            synchronized (regionEntry) {
              if (diskRegion.testIsRecoveredAndClear(regionEntry)) {
                wasRecovered = true;
                if (tmpValue == null) {
                  tmpValue = entry.isLocalInvalid() ? Token.LOCAL_INVALID : Token.INVALID;
                }

                // Compare the version stamps, and if they are equal
                // we can skip adding the entry we receive as part of GII.
                VersionStamp stamp = regionEntry.getVersionStamp();
                boolean entriesEqual = stamp != null && stamp.asVersionTag().equals(tag);

                // If the received entry and what we have in the cache
                // actually are equal, keep don't put the received
                // entry into the cache (this avoids writing a record to disk)
                if (entriesEqual) {
                  continue;
                }
                if (entry.isSerialized() && !Token.isInvalidOrRemoved(tmpValue)) {
                  tmpValue =
                      CachedDeserializableFactory.create((byte[]) tmpValue, region.getCache());
                }
                try {
                  if (tag != null) {
                    tag.replaceNullIDs(sender);
                  }
                  boolean record;
                  if (region.getVersionVector() != null && tag != null) {
                    region.getVersionVector().recordVersion(tag.getMemberID(), tag);
                    record = true;
                  } else {
                    record = (tmpValue != Token.TOMBSTONE);
                  }
                  if (record) {
                    this.entries.initialImagePut(entry.key, lastModified, tmpValue, wasRecovered,
                        true, tag, sender, isSynchronizing);
                    if (isSynchronizing) {
                      entriesToSynchronize.add(entry);
                    }
                  }
                } catch (RegionDestroyedException | CancelException e) {
                  return false;
                }
                didIIP = true;
              }
            }
            this.entries.lruUpdateCallback();
          }
        }

        if (keys != null) {
          if (tag == null) {
            keys.add(entry.key);
          } else {
            keys.add(entry.key + ",v=" + tag);
          }
        }
        if (!didIIP) {
          if (tmpValue == null) {
            tmpValue = entry.isLocalInvalid() ? Token.LOCAL_INVALID : Token.INVALID;
          } else if (entry.isSerialized()) {
            tmpValue = CachedDeserializableFactory.create((byte[]) tmpValue, region.getCache());
          }
          try {
            // null IDs in a version tag are meant to mean "this member", so
            // we need to change them to refer to the image provider.
            if (tag != null) {
              tag.replaceNullIDs(sender);
              if (tag.getMemberID() == null) {
                // It's possible that for persistent regions, a race condition between GII due to
                // region creation and a region destroy on the remote member can result in a
                // DiskVersionTag having a null memberID, so in that situation, return false to
                // allow the GII to be retried.
                if (isDebugEnabled) {
                  logger.debug("For Entry = {} with DiskVersionTag = {} from sender = {}, memberID"
                      + " was null; returning false from processChunk()", entry, tag, sender);
                }
                return false;
              }
            }
            if (isTraceEnabled) {
              logger.trace(
                  "processChunk:initialImagePut:key={},lastModified={},tmpValue={},wasRecovered={},tag={}",
                  entry.key, lastModified, tmpValue, wasRecovered, tag);
            }
            if (region.getVersionVector() != null && tag != null) {
              region.getVersionVector().recordVersion(tag.getMemberID(), tag);
            }
            this.entries.initialImagePut(entry.key, lastModified, tmpValue, wasRecovered, false,
                tag, sender, isSynchronizing);
            if (isSynchronizing) {
              entriesToSynchronize.add(entry);
            }
          } catch (RegionDestroyedException | CancelException e) {
            return false;
          }
        }
      }
      if (isSynchronizing && !entriesToSynchronize.isEmpty()) {
        LocalRegion owner = ((AbstractRegionMap) this.entries)._getOwner();
        LocalRegion region = owner instanceof BucketRegion ? owner.getPartitionedRegion() : owner;
        owner.getCache().invokeRegionEntrySynchronizationListenersAfterSynchronization(sender,
            region, entriesToSynchronize);
      }
      if (keys != null) {
        if (isDebugEnabled) {
          logger.debug("processed these initial image keys: {}", keys);
        }
      }
      if (internalBeforeCleanExpiredTombstones != null
          && internalBeforeCleanExpiredTombstones.getRegionName().equals(region.getName())) {
        internalBeforeCleanExpiredTombstones.run();
      }
      if (internalAfterSavedRVVEnd != null
          && internalAfterSavedRVVEnd.getRegionName().equals(region.getName())) {
        internalAfterSavedRVVEnd.run();
      }
      return true;
    } finally {
      if (diskRegion != null) {
        diskRegion.removeClearCountReference();
      }
    }
  }
Called from separate thread when reply is processed.

@param entries entries to add to the region
@return false if should abort (region was destroyed or cache was closed)
geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java