@Override
    protected void process(final ClusterDistributionManager dm) {
      final boolean isGiiDebugEnabled = logger.isTraceEnabled(LogMarker.INITIAL_IMAGE_VERBOSE);

      Throwable thr = null;
      final boolean lclAbortTest = abortTest;
      if (lclAbortTest) {
        abortTest = false;
      }
      DistributedRegion targetRegion = null;
      boolean sendFailureMessage = true;
      try {
        Assert.assertTrue(regionPath != null, "Region path is null.");
        final DistributedRegion rgn =
            (DistributedRegion) getGIIRegion(dm, regionPath, targetReinitialized);
        if (lostMemberID != null) {
          targetRegion = rgn;
        }
        if (rgn == null) {
          return;
        }

        // can simulate gc tombstone in middle of packing
        if (internalAfterReceivedRequestImage != null
            && internalAfterReceivedRequestImage.getRegionName().equals(rgn.getName())) {
          internalAfterReceivedRequestImage.run();
        }

        if (versionVector != null) {
          if (versionVector.isForSynchronization() && !rgn.getConcurrencyChecksEnabled()) {
            if (isGiiDebugEnabled) {
              logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                  "ignoring synchronization request as this region has no version vector");
            }
            replyNoData(dm, true, Collections.emptyMap());
            sendFailureMessage = false;
            return;
          }
          if (isGiiDebugEnabled) {
            logger.debug("checking version vector against region's ({})",
                rgn.getVersionVector().fullToString());
          }
          // [bruce] I suppose it's possible to have this check return a list of
          // specific versions that the sender is missing. The current check
          // just stops when it finds the first inconsistency
          if (!rgn.getVersionVector().isNewerThanOrCanFillExceptionsFor(versionVector)) {
            // Delta GII might have unfinished operations to send. Otherwise,
            // no need to send any data. This is a synchronization request and this region's
            // vector doesn't have anything that the other region needs
            if (unfinishedKeys == null || unfinishedKeys.isEmpty()) {
              if (isGiiDebugEnabled) {
                logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                    "version vector reports that I have nothing that the requester hasn't already seen");
              }
              replyNoData(dm, true, rgn.getVersionVector().getMemberToGCVersion());
              sendFailureMessage = false;
              return;
            }
          } else {
            if (isGiiDebugEnabled) {
              logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                  "version vector reports that I have updates the requester hasn't seen, remote rvv is {}",
                  versionVector);
            }
          }
        }

        final int numSeries = 1; // @todo ericz parallelize using series
        final int seriesNum = 0;

        // chunkEntries returns false if didn't finish
        if (isGiiDebugEnabled) {
          logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
              "RequestImageMessage: Starting chunkEntries for {}", rgn.getFullPath());
        }

        final InitialImageFlowControl flowControl =
            InitialImageFlowControl.register(dm, getSender());

        if (rgn instanceof HARegion) {
          ((HARegion) rgn).startServingGIIRequest();
        }
        boolean markedOngoingGII = false;
        try {
          boolean recoveringForLostMember = (lostMemberVersionID != null);
          RegionVersionHolder holderToSync = null;
          if (recoveringForLostMember && lostMemberID != null) {
            // wait for the lost member to be gone from this VM's membership and all ops applied to
            // the cache
            try {
              dm.getDistribution().waitForDeparture(lostMemberID);
              RegionVersionHolder rvh =
                  rgn.getVersionVector().getHolderForMember(lostMemberVersionID);
              if (rvh != null) {
                holderToSync = rvh.clone();
              }
              if (isGiiDebugEnabled) {
                RegionVersionHolder holderOfRequest =
                    versionVector.getHolderForMember(lostMemberVersionID);
                if (holderToSync != null
                    && holderToSync.isNewerThanOrCanFillExceptionsFor(holderOfRequest)) {
                  logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                      "synchronizeWith detected mismatch region version holder for lost member {}. Old is {}, new is {}",
                      lostMemberVersionID, holderOfRequest, holderToSync);
                }
              }
            } catch (TimeoutException e) {
              if (isGiiDebugEnabled) {
                logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                    "timed out waiting for the departure of {} before processing delta GII request",
                    lostMemberID);
              }
            }
          }
          if (rgn instanceof HARegion) {
            // long eventXferStart = System.currentTimeMillis();
            Map<? extends DataSerializable, ? extends DataSerializable> eventState =
                rgn.getEventState();
            if (eventState != null && eventState.size() > 0) {
              RegionStateMessage.send(dm, getSender(), processorId, eventState, true);
            }
          }
          if (checkTombstoneVersions && versionVector != null
              && rgn.getConcurrencyChecksEnabled()) {
            synchronized (rgn.getCache().getTombstoneService().getBlockGCLock()) {
              if (goWithFullGII(rgn, versionVector)) {
                if (isGiiDebugEnabled) {
                  logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE, "have to do fullGII");
                }
                versionVector = null; // full GII
              } else {
                // lock GIILock only for deltaGII
                int count = rgn.getCache().getTombstoneService().incrementGCBlockCount();
                markedOngoingGII = true;
                if (isGiiDebugEnabled) {
                  logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE, "There're {} Delta GII on going",
                      count);
                }
              }
            }
          }
          final RegionVersionHolder holderToSend = holderToSync;
          boolean finished = chunkEntries(rgn, CHUNK_SIZE_IN_BYTES, !keysOnly, versionVector,
              unfinishedKeys, flowControl, new ObjectIntProcedure() {
                int msgNum = 0;

                boolean last = false;

                /**
                 * @param entList ArrayList of entries
                 * @param b positive if last chunk
                 * @return true to continue to next chunk
                 */
                @Override
                public boolean executeWith(Object entList, int b) {
                  if (rgn.getCache().isClosed()) {
                    return false;
                  }

                  if (last) {
                    throw new InternalGemFireError("Already processed last chunk");
                  }

                  List<Entry> entries = uncheckedCast(entList);
                  // if abortTest, then never send last flag set to true
                  last = b > 0 && !lclAbortTest;
                  try {
                    boolean abort = rgn.isDestroyed();
                    if (!abort) {
                      int flowControlId = flowControl.getId();
                      Map<VersionSource<?>, Long> gcVersions = null;
                      if (last && rgn.getVersionVector() != null) {
                        gcVersions = rgn.getVersionVector().getMemberToGCVersion();
                      }
                      replyWithData(dm, entries, seriesNum, msgNum++, numSeries, last,
                          flowControlId, versionVector != null, holderToSend, gcVersions);
                    }
                    return !abort;
                  } catch (CancelException e) {
                    return false;
                  }
                }
              });


          if (isGiiDebugEnabled) {
            logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                "RequestImageMessage: ended chunkEntries for {}; finished = {}", rgn.getFullPath(),
                finished);
          }

          // Call to chunkEntries above will have sent at least one
          // reply with last==true for the last message. (unless doing abortTest or
          // region is destroyed or cache closed)
          if (finished && !lclAbortTest) {
            sendFailureMessage = false;
            return; // sent msg with last indicated
          }
          // One more chance to discover region or cache destruction...
          rgn.checkReadiness();
        } finally {
          if (markedOngoingGII) {
            int count = rgn.getCache().getTombstoneService().decrementGCBlockCount();
            assert count >= 0;
            if (count == 0) {
              markedOngoingGII = false;
              if (isGiiDebugEnabled) {
                logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE, "Delta GII count is reset");
              }
            }
          }
          if (rgn instanceof HARegion) {
            ((HARegion) rgn).endServingGIIRequest();
          }
          flowControl.unregister();
        }
        // This should never happen in production code!!!!

        // Code specific to abortTest... :-(
        Assert.assertTrue(lclAbortTest,
            this + ": Did not finish sending image, but region, cache, and DS are alive.");

        initiateLocalAbortForTest(dm);
      } catch (RegionDestroyedException e) {
        // thr = e; Don't marshal an exception here; just return null
        if (isGiiDebugEnabled) {
          logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
              "{}; Region destroyed: aborting image provision", this);
        }
      } catch (IllegalStateException e) {
        // thr = e; Don't marshal an exception here; just return null
        logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
            "{}; disk region deleted? aborting image provision", this, e);
      } catch (CancelException e) {
        // thr = e; Don't marshal an exception here; just return null
        if (isGiiDebugEnabled) {
          logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
              "{}; Cache Closed: aborting image provision", this);
        }
      } catch (VirtualMachineError err) {
        sendFailureMessage = false; // Don't try to respond!
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable t) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        thr = t;
      } finally {
        if (sendFailureMessage) {
          // if we get here then send reply possibly with an exception
          ReplyException rex = null;
          if (thr != null) {
            rex = new ReplyException(thr);
          }
          sendFailureMessage(dm, rex);
        } // !success

        if (lostMemberID != null && targetRegion != null) {
          if (lostMemberVersionID == null) {
            lostMemberVersionID = lostMemberID;
          }
          // check to see if the region in this cache needs to synchronize with others
          // it is possible that the cache is recover/restart of a member and not
          // scheduled to synchronize with others
          synchronizeIfNotScheduled(targetRegion, lostMemberID, lostMemberVersionID);
        }

        if (internalAfterSentImageReply != null
            && regionPath.endsWith(internalAfterSentImageReply.getRegionName())) {
          internalAfterSentImageReply.run();
        }
      }
    }
The versions in which this message was modified
geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java