/*
 * 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.tier.sockets;

;

import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_CREATE;
import static org.apache.geode.internal.cache.tier.MessageType.LOCAL_UPDATE;
import static org.apache.geode.internal.cache.tier.MessageType.REGISTER_DATASERIALIZERS;
import static org.apache.geode.internal.cache.tier.MessageType.REGISTER_INSTANTIATORS;
import static org.apache.geode.internal.cache.tier.MessageType.SERVER_TO_CLIENT_PING;

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.net.ssl.SSLException;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.StatisticDescriptor;
import org.apache.geode.Statistics;
import org.apache.geode.StatisticsType;
import org.apache.geode.StatisticsTypeFactory;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.InterestResultPolicy;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.client.ServerRefusedConnectionException;
import org.apache.geode.cache.client.SocketFactory;
import org.apache.geode.cache.client.internal.AuthenticateUserOp;
import org.apache.geode.cache.client.internal.ClientUpdater;
import org.apache.geode.cache.client.internal.Endpoint;
import org.apache.geode.cache.client.internal.EndpointManager;
import org.apache.geode.cache.client.internal.GetEventValueOp;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.QueueManager;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.DistributionStats;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.InternalDistributedSystem.DisconnectListener;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.tcpserver.HostAndPort;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.InternalInstantiator;
import org.apache.geode.internal.cache.ClientServerObserver;
import org.apache.geode.internal.cache.ClientServerObserverHolder;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.ClientSideHandshake;
import org.apache.geode.internal.cache.tier.InterestType;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.net.SocketCreator;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.internal.serialization.ByteArrayDataInput;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.statistics.StatisticsTypeFactoryImpl;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.PdxSerializationException;
import org.apache.geode.security.AuthenticationFailedException;
import org.apache.geode.security.AuthenticationRequiredException;
import org.apache.geode.security.GemFireSecurityException;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * {@code CacheClientUpdater} is a thread that processes update messages from a cache server and
 * {@linkplain org.apache.geode.cache.Region#localInvalidate(Object) invalidates} the local cache
 * based on the contents of those messages.
 *
 * @since GemFire 3.5
 */
public class CacheClientUpdater extends LoggingThread implements ClientUpdater, DisconnectListener {

  private static final Logger logger = LogService.getLogger();

  private static final int DEFAULT_SOCKET_BUFFER_SIZE = 32768;

  /**
   * true if the constructor successfully created a connection. If false, the run method for this
   * thread immediately exits.
   */
  private final boolean connected;

  /**
   * System of which we are a part
   */
  private final DistributedSystem system;

  /**
   * The socket by which we communicate with the server
   */
  private final Socket socket;

  /**
   * The output stream of the socket
   */
  private final OutputStream out;

  /**
   * The input stream of the socket
   */
  private final InputStream in;

  public ServerQueueStatus getServerQueueStatus() {
    return serverQueueStatus;
  }

  /**
   * server-side queue status at the time we connected to it
   */
  private ServerQueueStatus serverQueueStatus;

  /**
   * Failed updater from the endpoint previously known as the primary
   */
  private volatile ClientUpdater failedUpdater;

  /**
   * The buffer upon which we receive messages
   */
  private final ByteBuffer commBuffer;

  private boolean commBufferReleased; // TODO: fix synchronization

  private final CCUStats stats;

  /**
   * Cache for which we provide service TODO: lifecycle and synchronization need work
   */
  private /* final */ InternalCache cache;

  private /* final */ CachedRegionHelper cacheHelper;

  /**
   * Principle flag to signal thread's run loop to terminate
   */
  private final AtomicBoolean continueProcessing = new AtomicBoolean(true);

  /**
   * Is the client durable Used for bug 39010 fix
   */
  private final boolean isDurableClient;

  /**
   * Represents the server we are connected to
   */
  private final InternalDistributedMember serverId;

  /**
   * true if the EndPoint represented by this updater thread is primary
   */
  private final boolean isPrimary;

  /**
   * Added to avoid recording of the event if the concerned operation failed. See #43247
   */
  private boolean isOpCompleted;

  public static final String CLIENT_UPDATER_THREAD_NAME = "Cache Client Updater Thread ";

  /**
   * to enable test flag TODO: eliminate isUsedByTest
   */
  @MutableForTesting
  public static boolean isUsedByTest;

  /**
   * Indicates if full value was requested from server as a result of failure in applying delta
   * bytes. TODO: only used for test assertion
   */
  @MutableForTesting
  static boolean fullValueRequested = false;

  private final ServerLocation location;

  // TODO - remove these fields
  private final QueueManager qManager;
  private final EndpointManager eManager;
  private final Endpoint endpoint;

  private static final long MAX_CACHE_WAIT =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "CacheClientUpdater.MAX_WAIT", 120); // seconds

  /**
   * Return true if cache appears
   *
   * @return true if cache appears
   */
  private boolean waitForCache() {
    InternalCache cache;
    long tilt = System.currentTimeMillis() + MAX_CACHE_WAIT * 1000;
    for (;;) {
      if (quitting()) {
        logger.warn("{}: abandoned wait due to cancellation.", this);
        return false;
      }
      if (!connected) {
        logger.warn("{}: abandoned wait because it is no longer connected",
            this);
        return false;
      }
      if (System.currentTimeMillis() > tilt) {
        logger.warn("{}: wait timed out (more than {} seconds)",
            new Object[] {this, MAX_CACHE_WAIT});
        return false;
      }
      cache = GemFireCacheImpl.getInstance();
      if (cache != null && !cache.isClosed()) {
        break;
      }
      boolean interrupted = Thread.interrupted();
      try {
        Thread.sleep(1000);
      } catch (InterruptedException ignore) {
        interrupted = true;
      } finally {
        if (interrupted) {
          Thread.currentThread().interrupt();
        }
      }
    } // for
    this.cache = cache;
    cacheHelper = new CachedRegionHelper(cache);
    return true;
  }

  /**
   * Creates a new {@code CacheClientUpdater} with a given name that waits for a server to connect
   * on a given port.
   *
   * @param name descriptive name, used for our ThreadGroup
   * @param location the endpoint we represent
   * @param primary true if our endpoint is primary
   * @param ids the system we are distributing messages through
   *
   * @throws AuthenticationRequiredException when client is not configured to send credentials using
   *         security-* system properties but server expects credentials
   * @throws AuthenticationFailedException when authentication of the client fails
   * @throws ServerRefusedConnectionException when handshake fails for other reasons like using
   *         durable client ID that is already in use by another client or some server side
   *         exception while reading handshake/verifying credentials
   */
  public CacheClientUpdater(String name, ServerLocation location, boolean primary,
      DistributedSystem ids, ClientSideHandshake handshake, QueueManager qManager,
      EndpointManager eManager, Endpoint endpoint, int handshakeTimeout,
      SocketCreator socketCreator, SocketFactory socketFactory)
      throws AuthenticationRequiredException,
      AuthenticationFailedException, ServerRefusedConnectionException {
    this(name, location, primary, ids, handshake, qManager, eManager, endpoint, handshakeTimeout,
        socketCreator, new StatisticsProvider(), socketFactory);
  }

  /**
   * alternative constructor for unit tests. This constructor allows you to pass a
   * mock StatisticsProvider
   */
  public CacheClientUpdater(String name, ServerLocation location, boolean primary,
      DistributedSystem distributedSystem, ClientSideHandshake handshake, QueueManager qManager,
      EndpointManager eManager, Endpoint endpoint, int handshakeTimeout,
      SocketCreator socketCreator, StatisticsProvider statisticsProvider,
      SocketFactory socketFactory)
      throws AuthenticationRequiredException,
      AuthenticationFailedException, ServerRefusedConnectionException {
    super(name);
    system = distributedSystem;
    isDurableClient = handshake.isDurable();
    isPrimary = primary;
    this.location = location;
    this.qManager = qManager;
    // this holds the connection which this threads reads
    this.eManager = eManager;
    this.endpoint = endpoint;
    stats = statisticsProvider.createStatistics(distributedSystem, location);

    // Create the connection...
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      logger.debug("Creating asynchronous update connection");
    }

    boolean success = false;
    Socket mySock = null;
    InternalDistributedMember sid = null;
    ByteBuffer cb = null;
    OutputStream tmpOut = null;
    InputStream tmpIn = null;
    try {
      // Size of the server-to-client communication socket buffers
      int socketBufferSize =
          Integer.getInteger("BridgeServer.SOCKET_BUFFER_SIZE", DEFAULT_SOCKET_BUFFER_SIZE);

      mySock = socketCreator.forClient().connect(
          new HostAndPort(location.getHostName(), location.getPort()),
          handshakeTimeout, socketBufferSize, socketFactory::createSocket);
      mySock.setTcpNoDelay(true);
      mySock.setSendBufferSize(socketBufferSize);

      // Verify buffer sizes
      verifySocketBufferSize(socketBufferSize, mySock.getReceiveBufferSize(), "receive");
      verifySocketBufferSize(socketBufferSize, mySock.getSendBufferSize(), "send");

      // set the timeout for the handshake
      mySock.setSoTimeout(handshakeTimeout);
      tmpOut = mySock.getOutputStream();
      tmpIn = mySock.getInputStream();

      if (isDebugEnabled) {
        logger.debug(
            "Initialized server-to-client socket with send buffer size: {} bytes and receive buffer size: {} bytes",
            mySock.getSendBufferSize(), mySock.getReceiveBufferSize());
      }

      if (isDebugEnabled) {
        logger.debug(
            "Created connection from {}:{} to CacheClientNotifier on port {} for server-to-client communication",
            mySock.getInetAddress().getHostAddress(), mySock.getLocalPort(), mySock.getPort());
      }

      serverQueueStatus = handshake.handshakeWithSubscriptionFeed(mySock, isPrimary);
      if (serverQueueStatus.isPrimary() || serverQueueStatus.isNonRedundant()) {
        PoolImpl pool = (PoolImpl) this.qManager.getPool();
        if (!pool.getReadyForEventsCalled()) {
          pool.setPendingEventCount(serverQueueStatus.getServerQueueSize());
        }
      }

      int bufSize = 1024;
      try {
        bufSize = mySock.getSendBufferSize();
        if (bufSize < 1024) {
          bufSize = 1024;
        }
      } catch (SocketException ignore) {
      }
      cb = ServerConnection.allocateCommBuffer(bufSize, mySock);

      sid =
          new InternalDistributedMember(mySock.getInetAddress(), mySock.getPort(), false, true);

      success = true;
    } catch (ConnectException ignore) {
      if (!quitting()) {
        logger.warn("{} connection was refused", this);
      }
    } catch (SSLException ex) {
      if (!quitting()) {
        getSecurityLogger().warning(String.format("%s SSL negotiation failed. %s",
            this, ex));
        throw new AuthenticationFailedException(
            String.format("SSL negotiation failed with endpoint: %s",
                location),
            ex);
      }
    } catch (GemFireSecurityException ex) {
      if (!quitting()) {
        getSecurityLogger().warning(
            String.format(
                "%s: Security exception when creating server-to-client communication socket. %s",
                this, ex));
        throw ex;
      }
    } catch (IOException e) {
      if (!quitting()) {
        logger.warn(String.format(
            "%s: Caught following exception while attempting to create a server-to-client communication socket and will exit: %s",
            this, e),
            logger.isDebugEnabled() ? e : null);
      }
      eManager.serverCrashed(this.endpoint);
    } catch (ClassNotFoundException e) {
      if (!quitting()) {
        logger.warn("Unable to load the class: {}",
            e.getMessage());
      }
    } catch (ServerRefusedConnectionException e) {
      if (!quitting()) {
        logger.warn(String.format(
            "%s: Caught following exception while attempting to create a server-to-client communication socket and will exit: %s",
            this, e),
            logger.isDebugEnabled() ? e : null);
      }
      throw e;
    } finally {
      connected = success;
      socket = mySock;
      commBuffer = cb;
      out = tmpOut;
      in = tmpIn;
      serverId = sid;
      if (connected) {
        if (mySock != null) {
          try {
            mySock.setSoTimeout(0);
          } catch (SocketException ignore) {
            // ignore: nothing we can do about this
          }
        }
      } else {
        close();
      }
    }
  }

  private void releaseCommBuffer() {
    if (!commBufferReleased) {
      if (commBuffer != null) {
        synchronized (commBuffer) {
          if (!commBufferReleased) {
            commBufferReleased = true;
            ServerConnection.releaseCommBuffer(commBuffer);
          }
        }
      }
    }
  }

  public boolean isConnected() {
    return connected;
  }

  @Override
  public boolean isPrimary() {
    return isPrimary;
  }

  private InternalLogWriter getSecurityLogger() {
    return qManager.getSecurityLogger();
  }

  @Override
  public void setFailedUpdater(ClientUpdater failedUpdater) {
    this.failedUpdater = failedUpdater;
  }

  /**
   * Performs the work of the client update thread. Creates a {@code ServerSocket} and waits for the
   * server to connect to it.
   */
  @Override
  public void run() {
    EntryLogger.setSource(serverId, "RI");
    boolean addedListener = false;
    try {
      if (system instanceof InternalDistributedSystem) {
        ((InternalDistributedSystem) system).addDisconnectListener(this);
        addedListener = true;
      }

      if (!waitForCache()) {
        logger.warn("{}: no cache (exiting)", this);
        return;
      }
      processMessages();

    } catch (CancelException ignore) {
      // just bail

    } finally {
      if (addedListener) {
        ((InternalDistributedSystem) system).removeDisconnectListener(this);
      }
      close();
      EntryLogger.clearSource();
    }
  }

  /**
   * Notifies this thread to stop processing
   */
  private void stopProcessing() {
    continueProcessing.set(false);
  }

  /**
   * Stops the updater. It will wait for a while for the thread to finish to try to prevent
   * duplicates. Note: this method is not named stop because this is a Thread which has a deprecated
   * stop method.
   */
  private void stopUpdater() {
    boolean isSelfDestroying = Thread.currentThread() == this;
    stopProcessing();

    // need to also close the socket for this interrupt to wakeup
    // the thread. This fixes bug 35691.

    if (isAlive()) {
      if (logger.isDebugEnabled()) {
        logger.debug("{}: Stopping {}", location, this);
      }

      if (!isSelfDestroying) {
        interrupt();
        try {
          if (socket != null) {
            socket.close();
          }
        } catch (IOException e) {
          if (logger.isDebugEnabled()) {
            logger.debug(e.getMessage(), e);
          }
        }
      } // !isSelfDestroying
    } // isAlive
  }

  /**
   * Signals the run thread to stop, closes underlying resources.
   */
  @Override
  public void close() {
    continueProcessing.set(false); // signals we are done.

    // Close the socket. This will also cause the underlying streams to fail.
    try {
      if (socket != null) {
        socket.close();
      }
    } catch (IOException ignore) {
      // ignore
    }

    if (cacheHelper != null) {
      cacheHelper.close();
    }
    releaseCommBuffer();

    stats.close();
  }

  /**
   * Creates a cached {@link Message}object whose state is filled in with a message received from
   * the server.
   */
  private Message initializeMessage() {
    Message message = new Message(2, KnownVersion.CURRENT);
    message.setComms(socket, in, out, commBuffer, stats);
    return message;
  }

  /* refinement of method inherited from Thread */
  @Override
  public String toString() {
    return getName() + " (" + location.getHostName() + ':' + location.getPort() + ')';
  }

  /**
   * Handle a marker message
   *
   * @param clientMessage message containing the data
   */
  private void handleMarker(Message clientMessage) {
    try {
      final boolean isDebugEnabled = logger.isDebugEnabled();
      if (isDebugEnabled) {
        logger.debug("Received marker message of length ({} bytes)",
            clientMessage.getPayloadLength());
      }

      qManager.getState().processMarker();

      if (isDebugEnabled) {
        logger.debug("Processed marker message");
      }
    } catch (Exception e) {
      String message =
          "The following exception occurred while attempting to handle a marker.";
      handleException(message, e);
    }
  }

  /**
   * Create or update an entry
   *
   * @param clientMessage message containing the data
   */
  private void handleUpdate(Message clientMessage) {
    String regionName = null;
    Object key = null;
    Part valuePart = null;
    final boolean isDebugEnabled = logger.isDebugEnabled();

    try {
      isOpCompleted = false;

      // Retrieve the data from the put message parts
      if (isDebugEnabled) {
        logger.debug("Received put message of length ({} bytes)", clientMessage.getPayloadLength());
      }

      int partCnt = 0;
      Part regionNamePart = clientMessage.getPart(partCnt++);
      Part keyPart = clientMessage.getPart(partCnt++);
      boolean isDeltaSent = (Boolean) clientMessage.getPart(partCnt++).getObject();
      valuePart = clientMessage.getPart(partCnt++);
      Part callbackArgumentPart = clientMessage.getPart(partCnt++);
      VersionTag versionTag = (VersionTag) clientMessage.getPart(partCnt++).getObject();
      if (versionTag != null) {
        versionTag.replaceNullIDs((InternalDistributedMember) endpoint.getMemberId());
      }
      Part isInterestListPassedPart = clientMessage.getPart(partCnt++);
      Part hasCqsPart = clientMessage.getPart(partCnt++);

      EventID eventId =
          (EventID) clientMessage.getPart(clientMessage.getNumberOfParts() - 1).getObject();

      boolean withInterest = (Boolean) isInterestListPassedPart.getObject();
      boolean withCQs = (Boolean) hasCqsPart.getObject();

      regionName = regionNamePart.getCachedString();
      key = keyPart.getStringOrObject();
      Object callbackArgument = callbackArgumentPart.getObject();

      // Don't automatically deserialize the value.
      // Pass it onto the region as a byte[]. If it is a serialized
      // object, it will be stored as a CachedDeserializable and
      // deserialized only when requested.

      boolean isCreate = clientMessage.getMessageType() == LOCAL_CREATE;

      if (isDebugEnabled) {
        logger.debug(
            "Putting entry for region: {} key: {} create: {}{} callbackArgument: {} withInterest={} withCQs={} eventID={} version={}",
            regionName, key, isCreate,
            valuePart.isObject()
                ? new StringBuilder(" value: ").append(deserialize(valuePart.getSerializedForm()))
                : "",
            callbackArgument, withInterest, withCQs, eventId, versionTag);
      }

      LocalRegion region = (LocalRegion) cacheHelper.getRegion(regionName);

      byte[] deltaBytes = null;
      Object objectValue = null;
      boolean isValueObject;

      if (!isDeltaSent) {
        byte[] serializedForm = valuePart.getSerializedForm();

        if (!isCreate || !InternalDataSerializer.isSerializedNull(serializedForm)) {
          objectValue = valuePart.getSerializedForm();
        }

        if (withCQs) {
          objectValue = valuePart.getObject();
        }

        isValueObject = valuePart.isObject();
      } else {
        deltaBytes = valuePart.getSerializedForm();
        isValueObject = true;
      }

      if (region == null) {
        if (isDebugEnabled && !quitting()) {
          logger.debug("{}: Region named {} does not exist", this, regionName);
        }

      } else if (region.hasServerProxy() && ServerResponseMatrix
          .checkForValidStateAfterNotification(region, key, clientMessage.getMessageType())
          && (withInterest || !withCQs)) {
        @Released
        EntryEventImpl newEvent = null;

        try {
          // Create an event and put the entry
          newEvent = EntryEventImpl.create(region,
              clientMessage.getMessageType() == LOCAL_CREATE ? Operation.CREATE
                  : Operation.UPDATE,
              key, null /* newValue */, callbackArgument /* callbackArg */, true /* originRemote */,
              eventId.getDistributedMember());

          newEvent.setVersionTag(versionTag);
          newEvent.setFromServer(true);

          region.basicBridgeClientUpdate(eventId.getDistributedMember(), key, objectValue,
              deltaBytes,
              isValueObject, callbackArgument,
              clientMessage.getMessageType() == LOCAL_CREATE,
              qManager.getState().getProcessedMarker() || !isDurableClient, newEvent,
              eventId);

          isOpCompleted = true;

          // bug 45520 - ConcurrentCacheModificationException is not thrown and we must check this
          // flag
          if (withCQs && isDeltaSent) {
            objectValue = newEvent.getNewValue();
          }
        } catch (InvalidDeltaException ignore) {
          Part fullValuePart = requestFullValue(eventId, "Caught InvalidDeltaException.");
          region.getCachePerfStats().incDeltaFullValuesRequested();
          objectValue = fullValuePart.getObject();
          isValueObject = fullValuePart.isObject();

          region.basicBridgeClientUpdate(eventId.getDistributedMember(), key, objectValue, null,
              isValueObject, callbackArgument,
              clientMessage.getMessageType() == LOCAL_CREATE,
              qManager.getState().getProcessedMarker() || !isDurableClient, newEvent,
              eventId);

          isOpCompleted = true;
        } finally {
          if (newEvent != null) {
            newEvent.release();
          }
        }

        if (isDebugEnabled) {
          logger.debug("Put entry for region: {} key: {} callbackArgument: {}", regionName, key,
              callbackArgument);
        }
      }

      // Update CQs. CQs can exist without client region.
      if (withCQs) {
        Part numCqsPart = clientMessage.getPart(partCnt++);
        if (isDebugEnabled) {
          logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}",
              numCqsPart.getInt() / 2);
        }
        processCqs(clientMessage, partCnt, numCqsPart.getInt(),
            clientMessage.getMessageType(), key, objectValue, deltaBytes, eventId);
        isOpCompleted = true;
      }
    } catch (Exception e) {
      String message =
          String.format(
              "The following exception occurred while attempting to put entry (region: %s key: %s value: %s)",
              regionName, key, deserialize(valuePart.getSerializedForm()));
      handleException(message, e);
    }
  }

  private Part requestFullValue(EventID eventId, String reason) throws Exception {
    if (isUsedByTest) {
      fullValueRequested = true;
    }
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      logger.debug("{} Requesting full value...", reason);
    }
    Part result = (Part) GetEventValueOp.executeOnPrimary(qManager.getPool(), eventId, null);

    if (result == null) {
      // Just log a warning. Do not stop CCU thread.
      // TODO: throw a subclass of Exception
      throw new Exception("Could not retrieve full value for " + eventId);
    }

    if (isDebugEnabled) {
      logger.debug("Full value received.");
    }
    return result;
  }

  /**
   * Invalidate an entry
   *
   * @param clientMessage message describing the entry
   */
  private void handleInvalidate(Message clientMessage) {
    String regionName = null;
    Object key = null;
    final boolean isDebugEnabled = logger.isDebugEnabled();

    try {
      isOpCompleted = false;

      // Retrieve the data from the local-invalidate message parts
      if (isDebugEnabled) {
        logger.debug("Received invalidate message of length ({} bytes)",
            clientMessage.getPayloadLength());
      }

      int partCnt = 0;
      Part regionNamePart = clientMessage.getPart(partCnt++);
      Part keyPart = clientMessage.getPart(partCnt++);
      Part callbackArgumentPart = clientMessage.getPart(partCnt++);

      VersionTag versionTag = (VersionTag) clientMessage.getPart(partCnt++).getObject();
      if (versionTag != null) {
        versionTag.replaceNullIDs((InternalDistributedMember) endpoint.getMemberId());
      }

      Part isInterestListPassedPart = clientMessage.getPart(partCnt++);
      Part hasCqsPart = clientMessage.getPart(partCnt++);

      regionName = regionNamePart.getCachedString();
      key = keyPart.getStringOrObject();

      Object callbackArgument = callbackArgumentPart.getObject();
      boolean withInterest = (Boolean) isInterestListPassedPart.getObject();
      boolean withCQs = (Boolean) hasCqsPart.getObject();

      if (isDebugEnabled) {
        logger.debug(
            "Invalidating entry for region: {} key: {} callbackArgument: {} withInterest={} withCQs={} version={}",
            regionName, key, callbackArgument, withInterest, withCQs, versionTag);
      }

      LocalRegion region = (LocalRegion) cacheHelper.getRegion(regionName);
      if (region == null) {
        if (isDebugEnabled && !quitting()) {
          logger.debug("Region named {} does not exist", regionName);
        }

      } else {
        if (region.hasServerProxy() && (withInterest || !withCQs)) {
          try {
            Part eid = clientMessage.getPart(clientMessage.getNumberOfParts() - 1);
            EventID eventId = (EventID) eid.getObject();

            try {
              region.basicBridgeClientInvalidate(eventId.getDistributedMember(), key,
                  callbackArgument,
                  qManager.getState().getProcessedMarker() || !isDurableClient, eventId,
                  versionTag);
            } catch (ConcurrentCacheModificationException ignore) {
              // allow CQs to be processed
            }

            isOpCompleted = true;
            // fix for 36615
            qManager.getState().incrementInvalidatedStats();

            if (isDebugEnabled) {
              logger.debug("Invalidated entry for region: {} key: {} callbackArgument: {}",
                  regionName, key, callbackArgument);
            }
          } catch (EntryNotFoundException ignore) {
            if (isDebugEnabled && !quitting()) {
              logger.debug("Already invalidated entry for region: {} key: {} callbackArgument: {}",
                  regionName, key, callbackArgument);
            }
            isOpCompleted = true;
          }
        }
      }

      if (withCQs) {
        // The client may have been registered to receive invalidates for
        // create and updates operations. Get the actual region operation.
        Part regionOpType = clientMessage.getPart(partCnt++);
        Part numCqsPart = clientMessage.getPart(partCnt++);
        if (isDebugEnabled) {
          logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}",
              numCqsPart.getInt() / 2);
        }
        processCqs(clientMessage, partCnt, numCqsPart.getInt(), MessageType.valueOf(regionOpType.getInt()),
            key, null);
        isOpCompleted = true;
      }
    } catch (Exception e) {
      final String message =
          String.format(
              "The following exception occurred while attempting to invalidate entry (region: %s key: %s)",
              regionName, key);
      handleException(message, e);
    }
  }

  /**
   * locally destroy an entry
   *
   * @param clientMessage message describing the entry
   */
  private void handleDestroy(Message clientMessage) {
    String regionName = null;
    Object key = null;
    final boolean isDebugEnabled = logger.isDebugEnabled();

    try {
      isOpCompleted = false;
      // Retrieve the data from the local-destroy message parts
      if (isDebugEnabled) {
        logger.debug("Received destroy message of length ({} bytes)",
            clientMessage.getPayloadLength());
      }

      int partCnt = 0;
      Part regionNamePart = clientMessage.getPart(partCnt++);
      Part keyPart = clientMessage.getPart(partCnt++);
      Part callbackArgumentPart = clientMessage.getPart(partCnt++);

      VersionTag versionTag = (VersionTag) clientMessage.getPart(partCnt++).getObject();
