/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.geode.internal.cache;

import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.AFTER_INITIAL_IMAGE;
import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MessageWithReply;
import org.apache.geode.distributed.internal.OperationExecutors;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.NullDataOutputStream;
import org.apache.geode.internal.cache.InitialImageFlowControl.FlowControlPermitMessage;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.entries.DiskEntry;
import org.apache.geode.internal.cache.ha.HAContainerWrapper;
import org.apache.geode.internal.cache.persistence.DiskStoreID;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisor;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.DiskRegionVersionVector;
import org.apache.geode.internal.cache.versions.DiskVersionTag;
import org.apache.geode.internal.cache.versions.RegionVersionHolder;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.cache.vmotion.VMotionObserverHolder;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.internal.sequencelog.RegionLogger;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.serialization.StaticSerialization;
import org.apache.geode.internal.serialization.Versioning;
import org.apache.geode.internal.util.ObjectIntProcedure;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Handles requests for an initial image from a cache peer
 *
 */
public class InitialImageOperation {
  private static final Logger logger = LogService.getLogger();

  /**
   * internal flag used by unit tests to test early disconnect from distributed system
   */
  @MutableForTesting
  public static volatile boolean abortTest = false;

  /**
   * maximum number of bytes to put in a single message
   */
  @MutableForTesting
  public static int CHUNK_SIZE_IN_BYTES =
      Integer.getInteger("GetInitialImage.chunkSize", 500 * 1024);

  /**
   * Allowed number of in flight GII chunks
   */
  @MutableForTesting
  public static int CHUNK_PERMITS =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "GetInitialImage.CHUNK_PERMITS", 16);

  /**
   * maximum number of unfinished operations to be supported by delta GII
   */
  @MutableForTesting
  public static int MAXIMUM_UNFINISHED_OPERATIONS = Integer.getInteger(
      GeodeGlossary.GEMFIRE_PREFIX + "GetInitialImage.MAXIMUM_UNFINISHED_OPERATIONS", 10000);

  /**
   * Allowed number GIIs in parallel
   */
  @MutableForTesting
  public static final int MAX_PARALLEL_GIIS =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "GetInitialImage.MAX_PARALLEL_GIIS", 5);

  /**
   * the region we are fetching
   */
  protected final DistributedRegion region;

  /**
   * the underlying Map to receive our values
   */
  private final RegionMap entries;

  /**
   * true if we have received the last piece of the image
   */
  protected volatile boolean gotImage = false;

  /**
   * received region version holder for lost member, used by synchronizeWith only
   */
  protected RegionVersionHolder<?> rcvd_holderToSync;

  /**
   * received GC versions from the GCC source
   */
  protected Map<VersionSource<?>, Long> gcVersions;

  /**
   * true if this is delta gii
   */
  @MutableForTesting("Seems to be unused?")
  protected volatile boolean isDeltaGII = false;

  /**
   * for testing purposes
   */
  @MutableForTesting
  public static volatile int slowImageProcessing = 0;

  /**
   * for testing purposes
   */
  @MutableForTesting
  public static final AtomicInteger slowImageSleeps = new AtomicInteger();

  /**
   * for testing purposes
   */
  @MutableForTesting
  public static boolean VMOTION_DURING_GII = false;

  private boolean isSynchronizing;

  /** Creates a new instance of InitalImageOperation */
  InitialImageOperation(DistributedRegion region, RegionMap entries) {
    this.region = region;
    this.entries = entries;
  }

  /** a flag for inhibiting the use of StateFlushOperation before gii */
  private static final ThreadLocal<Boolean> inhibitStateFlush =
      ThreadLocal.withInitial(() -> Boolean.FALSE);

  /** inhibit use of StateFlush for the current thread */
  public static void setInhibitStateFlush(boolean inhibitIt) {
    inhibitStateFlush.set(inhibitIt);
  }

  public enum GIIStatus {
    NO_GII, GOTIMAGE_BY_FULLGII, GOTIMAGE_BY_DELTAGII;

    public static boolean didGII(GIIStatus giiStatus) {
      return (giiStatus == GIIStatus.GOTIMAGE_BY_FULLGII
          || giiStatus == GIIStatus.GOTIMAGE_BY_DELTAGII);
    }

    public static boolean didDeltaGII(GIIStatus giiStatus) {
      return giiStatus == GIIStatus.GOTIMAGE_BY_DELTAGII;
    }

    public static boolean didFullGII(GIIStatus giiStatus) {
      return giiStatus == GIIStatus.GOTIMAGE_BY_FULLGII;
    }
  }

  private GIIStatus reportGIIStatus() {
    if (!gotImage) {
      return GIIStatus.NO_GII;
    } else {
      // got image
      if (isDeltaGII) {
        return GIIStatus.GOTIMAGE_BY_DELTAGII;
      } else {
        return GIIStatus.GOTIMAGE_BY_FULLGII;
      }
    }
  }

  /**
   * Fetch an initial image from a single recipient
   *
   * @param recipientSet list of candidates to fetch from
   * @param targetReinitialized true if candidate should wait until initialized before responding
   * @param recoveredRVV recovered rvv
   * @return true if succeeded to get image
   * @throws org.apache.geode.cache.TimeoutException when it is unable to get a reply within the
   *         limit.
   */
  GIIStatus getFromOne(Set<InternalDistributedMember> recipientSet, boolean targetReinitialized,
      CacheDistributionAdvisor.InitialImageAdvice advice, boolean recoveredFromDisk,
      RegionVersionVector recoveredRVV) throws org.apache.geode.cache.TimeoutException {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    if (VMOTION_DURING_GII) {
      /*
       * TODO (ashetkar): recipientSet may contain more than one member. Ensure only the gii-source
       * member is vMotioned. The test hook may need to be placed at another point.
       */
      VMotionObserverHolder.getInstance().vMotionDuringGII(recipientSet, region);
    }
    // Make sure that candidates are regarded in random order
    ArrayList<InternalDistributedMember> recipients = new ArrayList<>(recipientSet);

    if (region.isUsedForSerialGatewaySenderQueue()) {
      AbstractGatewaySender sender = region.getSerialGatewaySender();
      if (sender != null) {
        InternalDistributedMember primary = sender.getSenderAdvisor().advisePrimaryGatewaySender();
        if (primary != null) {
          recipients.remove(primary);
          recipients.add(0, primary);
        }
      }
    } else {
      if (recipients.size() > 1) {
        Collections.shuffle(recipients);
      }
    }
    long giiStart = region.getCachePerfStats().startGetInitialImage();
    InternalDistributedMember provider = null;

    for (Iterator<InternalDistributedMember> itr = recipients.iterator(); !gotImage
        && itr.hasNext();) {
      // if we got a partial image from the previous recipient, then clear it

      InternalDistributedMember recipient = itr.next();
      provider = recipient;

      // In case of HARegion, before getting the region snapshot(image) get the filters
      // registered by the associated client and apply them.
      // As part of bug fix 39014, while creating the secondary HARegion/Queue, the
      // filters registered by the client is applied first and then the HARegion
      // initial image is applied. This is to process any events thats arriving while
      // GII is happening and is not part of the GII result.
      if (region instanceof HARegion) {
        try {
          // HARegion r = (HARegion)region;
          // if (!r.isPrimaryQueue()) {
          if (!requestFilterInfo(recipient)) {
            if (isDebugEnabled) {
              logger.debug("Failed to receive interest and CQ information from {}", recipient);
            }
          }
          // }
        } catch (Exception ex) {
          if (!itr.hasNext()) {
            if (isDebugEnabled) {
              logger.info("Failed while getting interest and CQ information from {}", recipient,
                  ex);
            }
          }
          continue;
        }
      }

      PersistenceAdvisor persistenceAdvisor = region.getPersistenceAdvisor();
      if (persistenceAdvisor != null) {
        try {
          persistenceAdvisor.updateMembershipView(recipient, targetReinitialized);
          persistenceAdvisor.setInitializing(region.getPersistentID());
        } catch (ReplyException e) {
          if (isDebugEnabled) {
            logger.debug("Failed to get membership view", e);
          }
          continue;
        }
      }
      final ClusterDistributionManager dm =
          (ClusterDistributionManager) region.getDistributionManager();
      final boolean allowDeltaGII = !FORCE_FULL_GII;
      Set<Object> keysOfUnfinishedOps = null;
      RegionVersionVector received_rvv = null;
      RegionVersionVector remote_rvv = null;
      if (region.getConcurrencyChecksEnabled()) {
        if (internalBeforeRequestRVV != null
            && internalBeforeRequestRVV.getRegionName().equals(region.getName())) {
          internalBeforeRequestRVV.run();
        }
        // Request the RVV from the provider and discover any operations on this
        // member that have not been performed on the provider.
        //
        // It is important that this happens *before* the state flush. An operation
        // maybe unfinished because a member crashed during distribution, or because
        // it is in flight right now. If it is in flight right now, we need to make
        // sure the provider receives the latest value for the operation before the
        // GII really starts.
        received_rvv = getRVVFromProvider(dm, recipient, targetReinitialized);
        if (received_rvv == null) {
          continue;
        }
        // remote_rvv will be filled with the versions of unfinished keys
        // then if recoveredRVV is still newer than the filled remote_rvv, do fullGII
        remote_rvv = received_rvv.getCloneForTransmission();
        keysOfUnfinishedOps = processReceivedRVV(remote_rvv, recoveredRVV, received_rvv);
        if (internalAfterCalculatedUnfinishedOps != null
            && internalAfterCalculatedUnfinishedOps.getRegionName().equals(region.getName())) {
          internalAfterCalculatedUnfinishedOps.run();
        }
        if (keysOfUnfinishedOps == null) {
          // if got rvv, keysOfUnfinishedOps at least will be empty
          continue;
        }
      }

      boolean inhibitFlush = inhibitStateFlush.get();
      if (!inhibitFlush && !region.doesNotDistribute()) {
        if (region instanceof BucketRegionQueue) {
          // get the corresponding userPRs and do state flush on all of them
          // TODO we should be able to do this state flush with a single
          // message, but that will require changing the messaging layer,
          // which has implications for a rolling upgrade.
          Collection<BucketRegion> userPRBuckets =
              ((BucketRegionQueue) (region)).getCorrespondingUserPRBuckets();
          if (isDebugEnabled) {
            logger.debug("The parent buckets of this shadowPR region are {}", userPRBuckets);
          }

          for (BucketRegion parentBucket : userPRBuckets) {
            if (isDebugEnabled) {
              logger.debug("Going to do state flush operation on the parent bucket.");
            }
            final StateFlushOperation sf;
            sf = new StateFlushOperation(parentBucket);
            final Set<InternalDistributedMember> r = new HashSet<>();
            r.addAll(advice.replicates);
            r.addAll(advice.preloaded);
            r.addAll(advice.others);
            r.addAll(advice.empties);
            r.addAll(advice.uninitialized);
            int processorType =
                targetReinitialized ? OperationExecutors.WAITING_POOL_EXECUTOR
                    : OperationExecutors.HIGH_PRIORITY_EXECUTOR;
            try {
              boolean success = sf.flush(r, recipient, processorType, true);
              if (!success) {
                continue;
              }
            } catch (InterruptedException ie) {
              Thread.currentThread().interrupt();
              region.getCancelCriterion().checkCancelInProgress(ie);
              return GIIStatus.NO_GII;
            }
            if (isDebugEnabled) {
              logger.debug("Completed state flush operation on the parent bucket.");
            }
          }
        }
        final StateFlushOperation sf;
        sf = new StateFlushOperation(region);
        final Set<InternalDistributedMember> r = new HashSet<>();
        r.addAll(advice.replicates);
        r.addAll(advice.preloaded);
        r.addAll(advice.others);
        r.addAll(advice.empties);
        r.addAll(advice.uninitialized);
        int processorType = targetReinitialized ? OperationExecutors.WAITING_POOL_EXECUTOR
            : OperationExecutors.HIGH_PRIORITY_EXECUTOR;
        try {
          boolean success = sf.flush(r, recipient, processorType, false);
          if (!success) {
            continue;
          }
        } catch (InterruptedException ie) {
          Thread.currentThread().interrupt();
          region.getCancelCriterion().checkCancelInProgress(ie);
          region.getCachePerfStats().endNoGIIDone(giiStart);
          return GIIStatus.NO_GII;
        }
      }

      RequestImageMessage m = new RequestImageMessage();
      m.regionPath = region.getFullPath();
      m.keysOnly = false;
      m.targetReinitialized = targetReinitialized;
      m.setRecipient(recipient);

      if (region.getConcurrencyChecksEnabled()) {
        if (allowDeltaGII && recoveredFromDisk) {
          if (!region.getDiskRegion().getRVVTrusted()) {
            if (isDebugEnabled) {
              logger.debug("Region {} recovered without EndGII flag, do full GII",
                  region.getFullPath());
            }
            m.versionVector = null;
          } else if (keysOfUnfinishedOps != null
              && keysOfUnfinishedOps.size() > MAXIMUM_UNFINISHED_OPERATIONS) {
            if (isDebugEnabled) {
              logger.debug(
                  "Region {} has {} unfinished operations, which exceeded threshold {}, do full GII instead",
                  region.getFullPath(), keysOfUnfinishedOps.size(),
                  MAXIMUM_UNFINISHED_OPERATIONS);
            }
            m.versionVector = null;

          } else if (region.getDiskStore() != null && region.getDiskStore().isDataCorrupted()) {
            if (isDebugEnabled) {
              logger.debug("Data corrupted, do full GII instead");
            }
            m.versionVector = null;
          } else {
            if (recoveredRVV.isNewerThanOrCanFillExceptionsFor(remote_rvv)) {
              m.versionVector = null;
              if (isDebugEnabled) {
                logger.debug(
                    "Region {}: after filled versions of unfinished keys, recovered rvv is still newer than remote rvv:{}. recovered rvv is {}. Do full GII",
                    region.getFullPath(), remote_rvv, recoveredRVV);
              }
            } else {
              m.versionVector = recoveredRVV;
              m.unfinishedKeys = keysOfUnfinishedOps;
              if (isDebugEnabled) {
                logger.debug(
                    "Region {} recovered with EndGII flag, rvv is {}. Received rvv is {}. Do delta GII",
                    region.getFullPath(), m.versionVector, received_rvv);
              }
            }
          }
          m.checkTombstoneVersions = true;
        }
        if (received_rvv != null) {
          // pack the original RVV, then save the received one
          if (internalBeforeSavedReceivedRVV != null
              && internalBeforeSavedReceivedRVV.getRegionName().equals(region.getName())) {
            internalBeforeSavedReceivedRVV.run();
          }
          saveReceivedRVV(received_rvv);
          if (internalAfterSavedReceivedRVV != null
              && internalAfterSavedReceivedRVV.getRegionName().equals(region.getName())) {
            internalAfterSavedReceivedRVV.run();
          }
        }
      }

      ImageProcessor processor = new ImageProcessor(region.getSystem(), recipient);
      dm.acquireGIIPermitUninterruptibly();
      try {
        m.processorId = processor.getProcessorId();
        if (region.isUsedForPartitionedRegionBucket()
            && region.getDistributionConfig().getAckSevereAlertThreshold() > 0) {
          processor.enableSevereAlertProcessing();
          m.severeAlertEnabled = true;
        }

        // do not remove the following log statement
        logger.info("Region {} requesting initial image from {}, recovered RVV is {}",
            new Object[] {region.getName(), recipient, recoveredRVV});

        dm.putOutgoing(m);
        region.cache.getCancelCriterion().checkCancelInProgress(null);
        if (internalAfterSentRequestImage != null
            && internalAfterSentRequestImage.getRegionName().equals(region.getName())) {
          internalAfterSentRequestImage.run();
        }
        try {
          processor.waitForRepliesUninterruptibly();

          // review unfinished keys and remove untouched entries
          if (region.getDataPolicy().withPersistence() && keysOfUnfinishedOps != null
              && !keysOfUnfinishedOps.isEmpty()) {
            final DiskRegion dr = region.getDiskRegion();
            assert dr != null;
            for (Object key : keysOfUnfinishedOps) {
              RegionEntry re = entries.getEntry(key);
              if (re == null) {
                continue;
              }
              if (logger.isTraceEnabled(LogMarker.INITIAL_IMAGE_VERBOSE)) {
                logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
                    "Processing unfinished operation:entry={}", re);
              }
              DiskEntry de = (DiskEntry) re;
              synchronized (de) {
                DiskId id = de.getDiskId();
                if (id != null && EntryBits.isRecoveredFromDisk(id.getUserBits())) {
                  region.destroyRecoveredEntry(key);
                  if (isDebugEnabled) {
                    logger.debug("Deleted unfinished keys:key={}", key);
                  }
                }
              }
            }
          }

          continue;
        } catch (InternalGemFireException ex) {
          Throwable cause = ex.getCause();
          if (cause instanceof org.apache.geode.cache.TimeoutException) {
            throw (org.apache.geode.cache.TimeoutException) cause;
          }
          throw ex;
        } catch (ReplyException e) {
          if (!region.isDestroyed()) {
            e.handleCause();
          }
        } finally {
          ImageState imgState = region.getImageState();

          if (imgState.getClearRegionFlag()) {
            // Asif : Since the operation has been completed clear flag
            imgState.setClearRegionFlag(false, null);
          }

          // Make sure we have applied the tombstone GC as seen on the GII
          // source
          if (gcVersions != null) {
            region.getGemFireCache().getTombstoneService().gcTombstones(region, gcVersions,
                false);
          }

          if (gotImage) {
            RegionLogger.logGII(region.getFullPath(), recipient,
                region.getDistributionManager().getDistributionManagerId(),
                region.getPersistentID());
          }
          if (gotImage) {
            // TODO add localizedString
            logger.info("{} is done getting image from {}. isDeltaGII is {}", region.getName(),
                recipient, isDeltaGII);
          } else {
            // TODO add localizedString
            logger.info("{} failed to get image from {}", region.getName(), recipient);
          }
          if (region.getDataPolicy().withPersistence()) {
            logger.info("Region {} initialized persistent id: {} with data from {}.",
                new Object[] {region.getName(), region.getPersistentID(),
                    recipient});
          }
          // bug 39050 - no partial images after GII when network partition
          // detection is enabled
          if (!gotImage) {
            region.cleanUpAfterFailedGII(recoveredFromDisk);
          } else if (received_rvv != null) {
            checkForUnrecordedOperations();
          }
        }
      } finally {
        dm.releaseGIIPermit();
        processor.cleanup();
      }
    } // for

    if (gotImage) {
      region.recordEventStateFromImageProvider(provider);
      region.getCachePerfStats().endGetInitialImage(giiStart);
      if (isDeltaGII) {
        region.getCachePerfStats().incDeltaGIICompleted();
      }
    } else {
      region.getCachePerfStats().endNoGIIDone(giiStart);
    }
    return reportGIIStatus();
  }

  /**
   * synchronize with another member (delta GII from it). If lostMember is not null, then only
   * changes that it made to the image provider will be sent back. Otherwise all changes made to the
   * image provider will be compared with those made to this cache and a full delta will be sent.
   */
  public void synchronizeWith(InternalDistributedMember target, VersionSource lostMemberVersionID,
      InternalDistributedMember lostMember) {
    final ClusterDistributionManager dm =
        (ClusterDistributionManager) region.getDistributionManager();

    isSynchronizing = true;
    RequestImageMessage m = new RequestImageMessage();
    m.regionPath = region.getFullPath();
    m.keysOnly = false;

    if (lostMemberVersionID != null) {
      m.versionVector = region.getVersionVector().getCloneForTransmission(lostMemberVersionID);
      m.lostMemberVersionID = lostMemberVersionID;
      m.lostMemberID = lostMember;
    } else {
      m.versionVector = region.getVersionVector().getCloneForTransmission();
    }
    m.setRecipient(target);
    ImageProcessor processor = new ImageProcessor(region.getSystem(), target);
    dm.acquireGIIPermitUninterruptibly();
    try {
      m.processorId = processor.getProcessorId();
      if (region.isUsedForPartitionedRegionBucket()
          && region.getDistributionConfig().getAckSevereAlertThreshold() > 0) {
        processor.enableSevereAlertProcessing();
        m.severeAlertEnabled = true;
      }

      logger.info("Region {} is requesting synchronization with {} for {}", region.getName(),
          target, lostMember);

      long hisVersion = region.getVersionVector().getVersionForMember(lostMemberVersionID);

      dm.putOutgoing(m);
      region.cache.getCancelCriterion().checkCancelInProgress(null);
      try {
        processor.waitForRepliesUninterruptibly();
        ImageState imgState = region.getImageState();
        if (imgState.getClearRegionFlag()) {
          imgState.setClearRegionFlag(false, null);
        }
      } catch (InternalGemFireException ex) {
        Throwable cause = ex.getCause();
        if (cause instanceof org.apache.geode.cache.TimeoutException) {
          throw (org.apache.geode.cache.TimeoutException) cause;
        }
        throw ex;
      } catch (ReplyException e) {
        if (!region.isDestroyed()) {
          e.handleCause();
        }
      } finally {
        if (gotImage) {
          region.getVersionVector().removeExceptionsFor(target, hisVersion);
          RegionVersionHolder holder =
              region.getVersionVector().getHolderForMember(lostMemberVersionID);
          if (rcvd_holderToSync != null
              && rcvd_holderToSync.isNewerThanOrCanFillExceptionsFor(holder)) {
            logger.info(
                "synchronizeWith detected mismatch region version holder for lost member {}. Old is {}, new is {}",
                lostMemberVersionID, holder, rcvd_holderToSync);
            region.getVersionVector().initializeVersionHolder(lostMemberVersionID,
                rcvd_holderToSync);
          }
          RegionLogger.logGII(region.getFullPath(), target,
              region.getDistributionManager().getDistributionManagerId(), region.getPersistentID());
        }
        if (gotImage) {
          if (logger.isDebugEnabled()) {
            logger.debug("{} is done synchronizing with {}", region.getName(), target);
          }
        } else {
          if (logger.isDebugEnabled()) {
            logger.debug(
                "{} received no synchronization data from {} which could mean that we are already synchronized",
                region.getName(), target);
          }
        }
      }
    } finally {
      dm.releaseGIIPermit();
      processor.cleanup();
    }
  }

  private void checkForUnrecordedOperations() {
    final boolean isTraceEnabled = logger.isTraceEnabled();

    // bug #48962 - a change could have been received from a member
    // that the image provider didn't see. This can happen if the
    // image provider is creating the region in parallel with this member.
    // We have to check all of the received versions for members that
    // left during GII to see if the RVV contains them.
    RegionVersionVector rvv = region.getVersionVector();
    if (region.getConcurrencyChecksEnabled() && rvv != null) {
      ImageState state = region.getImageState();
      if (state.hasLeftMembers()) {
        Set<VersionSource> needsSync = null;
        Set<VersionSource> leftMembers = state.getLeftMembers();
        Iterator<ImageState.VersionTagEntry> tags = state.getVersionTags();
        while (tags.hasNext()) {
          ImageState.VersionTagEntry tag = tags.next();
          if (isTraceEnabled) {
            logger.trace("checkForUnrecordedOperations: processing tag {}", tag);
          }
          if (leftMembers.contains(tag.getMemberID())
              && !rvv.contains(tag.getMemberID(), tag.getRegionVersion())) {
            if (needsSync == null) {
              needsSync = new HashSet<>();
            }
            needsSync.add(tag.getMemberID());
            rvv.recordVersion(tag.getMemberID(), tag.getRegionVersion());
          }
        }
        if (needsSync != null) {
          // we need to tell the image provider to request syncs on the given
          // member(s) These will either be DistributedMember IDs or DiskStore IDs
          RequestSyncMessage msg = new RequestSyncMessage();
          msg.regionPath = region.getFullPath();
          msg.lostVersionSources = needsSync.toArray(new VersionSource[0]);
          Set<InternalDistributedMember> recipients =
              region.getCacheDistributionAdvisor().adviseReplicates();
          if (!recipients.isEmpty()) {
            msg.setRecipients(recipients);
            if (logger.isDebugEnabled()) {
              logger.debug("Local versions were found that the image provider has not seen for {}",
                  needsSync);
            }
            region.getDistributionManager().putOutgoing(msg);
          }
        }
      }
    }
  }

  /**
   * test hook invokation
   */
  public static void beforeGetInitialImage(DistributedRegion region) {
    if (internalBeforeGetInitialImage != null
        && internalBeforeGetInitialImage.getRegionName().equals(region.getName())) {
      internalBeforeGetInitialImage.run();
    }

  }


  /**
   * transfer interest/cq registrations from the image provider to this VM
   *
   * @return whether the operation succeeded in transferring anything
   */
  private boolean requestFilterInfo(final InternalDistributedMember recipient) {
    // Request for Filter Information before getting the HARegion snapshot.
    final DistributionManager dm = region.getDistributionManager();
    final RequestFilterInfoMessage filterInfoMsg = new RequestFilterInfoMessage();
    filterInfoMsg.regionPath = region.getFullPath();
    filterInfoMsg.setRecipient(recipient);
    final FilterInfoProcessor processor = new FilterInfoProcessor(region.getSystem(), recipient);
    filterInfoMsg.processorId = processor.getProcessorId();
    dm.putOutgoing(filterInfoMsg);

    try {
      processor.waitForRepliesUninterruptibly();
      return processor.filtersReceived;
    } catch (InternalGemFireException ex) {
      final Throwable cause = ex.getCause();
      if (cause instanceof org.apache.geode.cache.TimeoutException) {
        throw (org.apache.geode.cache.TimeoutException) cause;
      }
      throw ex;
    } catch (ReplyException e) {
      if (!region.isDestroyed()) {
        e.handleCause();
