void registerClientInternal(final ClientRegistrationMetadata clientRegistrationMetadata,
      final Socket socket,
      final boolean isPrimary,
      final long acceptorId, final boolean notifyBySubscription)
      throws IOException, CacheException, ClassNotFoundException, NoSuchMethodException,
      InvocationTargetException, IllegalAccessException {
    ClientProxyMembershipID clientProxyMembershipID =
        clientRegistrationMetadata.getClientProxyMembershipID();
    byte clientConflation = clientRegistrationMetadata.getClientConflation();
    KnownVersion clientVersion = clientRegistrationMetadata.getClientVersion();

    CacheClientProxy cacheClientProxy = getClientProxy(clientProxyMembershipID);
    DistributedMember member = clientProxyMembershipID.getDistributedMember();
    DistributedSystem system = getCache().getDistributedSystem();
    Properties sysProps = system.getProperties();
    String authenticator = sysProps.getProperty(SECURITY_CLIENT_AUTHENTICATOR);
    // in multi-user case, this would return null
    Object subjectOrPrincipal =
        getSubjectOrPrincipal(clientRegistrationMetadata, member, system, authenticator);
    Subject subject = subjectOrPrincipal instanceof Subject ? (Subject) subjectOrPrincipal : null;

    // Initialize the socket
    socket.setTcpNoDelay(true);
    socket.setSendBufferSize(CacheClientNotifier.socketBufferSize);
    socket.setReceiveBufferSize(CacheClientNotifier.socketBufferSize);


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

    // Determine whether the client is durable or not.
    byte responseByte = CommunicationMode.SuccessfulServerToClient.getModeNumber();
    boolean successful = true;
    boolean clientIsDurable = clientProxyMembershipID.isDurable();
    if (logger.isDebugEnabled()) {
      if (clientIsDurable) {
        logger.debug("CacheClientNotifier: Attempting to register durable client: {}",
            clientProxyMembershipID.getDurableId());
      } else {
        logger.debug("CacheClientNotifier: Attempting to register non-durable client");
      }
    }

    byte endpointType = 0x00;
    int queueSize = 0;
    String unsuccessfulMsg = null;
    if (clientIsDurable) {
      if (cacheClientProxy == null) {
        if (isTimedOut(clientProxyMembershipID)) {
          queueSize = PoolImpl.PRIMARY_QUEUE_TIMED_OUT;
        } else {
          queueSize = PoolImpl.PRIMARY_QUEUE_NOT_AVAILABLE;
        }
        // No proxy exists for this durable client. It must be created.
        if (logger.isDebugEnabled()) {
          logger.debug(
              "CacheClientNotifier: No proxy exists for durable client with id {}. It must be created.",
              clientProxyMembershipID.getDurableId());
        }
        cacheClientProxy =
            cacheClientProxyFactory.create(this, socket, clientProxyMembershipID, isPrimary,
                clientConflation, clientVersion, acceptorId, notifyBySubscription,
                cache.getSecurityService(), subject, statisticsClock);
        successful = initializeProxy(cacheClientProxy);
      } else {
        cacheClientProxy.setSubject(subject);
        if (cacheClientProxy.isPrimary()) {
          endpointType = (byte) 2;
        } else {
          endpointType = (byte) 1;
        }
        queueSize = cacheClientProxy.getQueueSize();
        // A proxy exists for this durable client. It must be reinitialized.
        if (cacheClientProxy.isPaused()) {
          if (CacheClientProxy.testHook != null) {
            CacheClientProxy.testHook.doTestHook("CLIENT_PRE_RECONNECT");
          }
          if (cacheClientProxy.lockDrain()) {
            try {
              if (logger.isDebugEnabled()) {
                logger.debug(
                    "CacheClientNotifier: A proxy exists for durable client with id {}. This proxy will be reinitialized: {}",
                    clientProxyMembershipID.getDurableId(), cacheClientProxy);
              }
              statistics.incDurableReconnectionCount();
              cacheClientProxy.getProxyID()
                  .updateDurableTimeout(clientProxyMembershipID.getDurableTimeout());
              cacheClientProxy.reinitialize(socket, clientProxyMembershipID,
                  isPrimary, clientConflation,
                  clientVersion);
              cacheClientProxy.setMarkerEnqueued(true);
              if (CacheClientProxy.testHook != null) {
                CacheClientProxy.testHook.doTestHook("CLIENT_RECONNECTED");
              }
            } finally {
              cacheClientProxy.unlockDrain();
            }
          } else {
            unsuccessfulMsg =
                "CacheClientNotifier: Connection refused due to cq queue being drained from admin command, please wait...";
            logger.warn(unsuccessfulMsg);
            responseByte = Handshake.REPLY_REFUSED;
            if (CacheClientProxy.testHook != null) {
              CacheClientProxy.testHook.doTestHook("CLIENT_REJECTED_DUE_TO_CQ_BEING_DRAINED");
            }
          }
        } else {
          // The existing proxy is already running (which means that another
          // client is already using this durable id.
          unsuccessfulMsg =
              String.format(
                  "The requested durable client has the same identifier ( %s ) as an existing durable client ( %s ). Duplicate durable clients are not allowed.",
                  clientProxyMembershipID.getDurableId(), cacheClientProxy);
          logger.warn(unsuccessfulMsg);
          // Set the unsuccessful response byte.
          responseByte = Handshake.REPLY_EXCEPTION_DUPLICATE_DURABLE_CLIENT;
        }
      }
    } else {
      CacheClientProxy staleClientProxy = getClientProxy(clientProxyMembershipID);
      if (staleClientProxy != null) {
        // A proxy exists for this non-durable client. It must be closed.
        if (logger.isDebugEnabled()) {
          logger.debug(
              "CacheClientNotifier: A proxy exists for this non-durable client. It must be closed.");
        }
        if (staleClientProxy.startRemoval()) {
          staleClientProxy.waitRemoval();
        } else {
          staleClientProxy.close(false, false); // do not check for queue, just close it
          removeClientProxy(staleClientProxy); // remove old proxy from proxy set
        }
      } // non-null stale proxy

      // Create the new proxy for this non-durable client
      cacheClientProxy =
          new CacheClientProxy(this, socket, clientProxyMembershipID, isPrimary, clientConflation,
              clientVersion, acceptorId, notifyBySubscription, cache.getSecurityService(), subject,
              statisticsClock);
      successful = initializeProxy(cacheClientProxy);
    }

    if (!successful) {
      cacheClientProxy = null;
      responseByte = Handshake.REPLY_REFUSED;
      unsuccessfulMsg =
          String.format(
              "A previous connection attempt from this client is still being processed: %s",
              clientProxyMembershipID);
      logger.warn(unsuccessfulMsg);
    } else if (subject != null) {
      secureLogger.debug("CacheClientProxy {} using subject {} {} ", cacheClientProxy,
          subject.getPrincipal(), subject);
    }

    // Tell the client that the proxy has been registered using the response
    // byte. This byte will be read on the client by the CacheClientUpdater to
    // determine whether the registration was successful. The times when
    // registration is unsuccessful currently are if a duplicate durable client
    // is attempted to be registered or authentication fails.
    try {
      DataOutputStream dos =
          new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
      // write the message type, message length and the error message (if any)
      socketMessageWriter.writeHandshakeMessage(dos, responseByte, unsuccessfulMsg, clientVersion,
          endpointType, queueSize);
    } catch (IOException ioe) {// remove the added proxy if we get IOException.
      if (cacheClientProxy != null) {
        // do not check for queue, just close it
        boolean keepProxy = cacheClientProxy.close(false, false);
        if (!keepProxy) {
          removeClientProxy(cacheClientProxy);
        }
      }
      throw ioe;
    }

    if (unsuccessfulMsg != null && logger.isDebugEnabled()) {
      logger.debug(unsuccessfulMsg);
    }

    // If the client is not durable, start its message processor
    // Starting it here (instead of in the CacheClientProxy constructor)
    // will ensure that the response byte is sent to the client before
    // the marker message. If the client is durable, the message processor
    // is not started until the clientReady message is received.
    if (!clientIsDurable && cacheClientProxy != null
        && responseByte == CommunicationMode.SuccessfulServerToClient.getModeNumber()) {
      // The startOrResumeMessageDispatcher tests if the proxy is a primary.
      // If this is a secondary proxy, the dispatcher is not started.
      // The false parameter signifies that a marker message has not already been
      // processed. This will generate and send one.
      cacheClientProxy.startOrResumeMessageDispatcher(false);
    }

    if (responseByte == CommunicationMode.SuccessfulServerToClient.getModeNumber()) {
      if (logger.isDebugEnabled()) {
        logger.debug("CacheClientNotifier: Successfully registered {}", cacheClientProxy);
      }
      performPostAuthorization(cacheClientProxy, clientProxyMembershipID, member,
          sysProps,
          subjectOrPrincipal);
    } else {
      try {
        // prevent leak by closing socket
        socket.close();
      } catch (IOException ignore) {
      }
      logger.warn(
          "CacheClientNotifier: Unsuccessfully registered client with identifier {} and response code {}",
          new Object[] {clientProxyMembershipID, responseByte});
    }
  }
Continues the registration of a new client that wants to receive updates with this server.

@param clientRegistrationMetadata Contains registration info pertaining to the client
@param socket The socket over which the server communicates with the client.
@param isPrimary Whether server is the primary subscription end point for this client
@param acceptorId ID of the acceptor used to clean up the client connection
@param notifyBySubscription Whether the Server is running in NotifyBySubscription mode
@throws IOException Can occur if there are issues communicating over the socket
@throws CacheException A generic exception, which indicates a cache error has occurred.
@throws ClassNotFoundException Thrown when the ClientProxyMembershipID class is not found
@throws NoSuchMethodException Potentially thrown by performPostAuthorization
@throws InvocationTargetException Potentially thrown by performPostAuthorization
@throws IllegalAccessException Potentially thrown by performPostAuthorization
geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java