static void handleException(final @NotNull Throwable throwable,
      final @NotNull Connection connection,
      final int retryCount, final boolean finalAttempt,
      final boolean timeoutFatal, final @NotNull CancelCriterion cancelCriterion,
      final @NotNull EndpointManager endpointManager) throws CacheRuntimeException {

    cancelCriterion.checkCancelInProgress(throwable);

    if (logger.isDebugEnabled() && !(throwable instanceof java.io.EOFException)) {
      if (throwable instanceof java.net.SocketTimeoutException) {
        logger.debug("OpExecutor.handleException on Connection to {} read timed out",
            connection.getServer());
      } else {
        logger.debug("OpExecutor.handleException on Connection to {}", connection.getServer(),
            throwable);
      }
    }

    // first take care of all exceptions that should not invalidate the
    // connection and do not need to be logged

    GemFireException exceptionToThrow = null;
    String title;
    boolean invalidateServer = true;
    boolean warn = true;
    boolean forceThrow = false;
    Throwable cause = throwable;

    if (throwable instanceof MessageTooLargeException) {
      title = null;
      exceptionToThrow = new GemFireIOException("message is too large to transmit", throwable);
    } else if (throwable instanceof NotSerializableException) {
      title = null; // no message
      exceptionToThrow = new SerializationException("Pool message failure", throwable);
    } else if (throwable instanceof BatchException || throwable instanceof BatchException70) {
      title = null; // no message
      exceptionToThrow = new ServerOperationException(throwable);
    } else if (throwable instanceof RegionDestroyedException) {
      invalidateServer = false;
      title = null;
      exceptionToThrow = (RegionDestroyedException) throwable;
    } else if (throwable instanceof GemFireSecurityException) {
      title = null;
      exceptionToThrow = new ServerOperationException(throwable);
    } else if (throwable instanceof SerializationException) {
      title = null; // no message
      exceptionToThrow = new ServerOperationException(throwable);
    } else if (throwable instanceof CopyException) {
      title = null; // no message
      exceptionToThrow = new ServerOperationException(throwable);
    } else if (throwable instanceof ClassNotFoundException) {
      title = null; // no message
      exceptionToThrow = new ServerOperationException(throwable);
    } else if (throwable instanceof TransactionException) {
      title = null; // no message
      exceptionToThrow = (TransactionException) throwable;
      invalidateServer = false;
    } else if (throwable instanceof SynchronizationCommitConflictException) {
      title = null;
      exceptionToThrow = (SynchronizationCommitConflictException) throwable;
      invalidateServer = false;
    } else if (throwable instanceof SocketException) {
      if ("Socket closed".equals(throwable.getMessage())
          || "Connection reset".equals(throwable.getMessage())
          || "Connection refused: connect".equals(throwable.getMessage())
          || "Connection refused".equals(throwable.getMessage())) {
        title = throwable.getMessage();
      } else {
        title = "SocketException";
      }
    } else if (throwable instanceof SocketTimeoutException) {
      invalidateServer = timeoutFatal;
      title = "socket timed out on client";
      // specific cause checks will fail if cause is not null here
      cause = null;
    } else if (throwable instanceof ConnectionDestroyedException) {
      invalidateServer = false;
      title = "connection was asynchronously destroyed";
      // specific cause checks will fail if cause is not null here
      cause = null;
    } else if (throwable instanceof java.io.EOFException) {
      title = "closed socket on server";
    } else if (throwable instanceof IOException) {
      title = "IOException";
    } else if (throwable instanceof BufferUnderflowException) {
      title = "buffer underflow reading from server";
    } else if (throwable instanceof CancelException) {
      title = "Cancelled";
      warn = false;
    } else if (throwable instanceof InternalFunctionInvocationTargetException) {
      // In this case, function will be re executed
      title = null;
      exceptionToThrow = (InternalFunctionInvocationTargetException) throwable;
    } else if (throwable instanceof FunctionInvocationTargetException) {
      // in this case function will not be re executed
      title = null;
      exceptionToThrow = (GemFireException) throwable;
    } else if (throwable instanceof PutAllPartialResultException) {
      title = null;
      exceptionToThrow = (PutAllPartialResultException) throwable;
      invalidateServer = false;
    } else {
      Throwable t = throwable.getCause();
      if ((t instanceof IOException)
          || (t instanceof SerializationException) || (t instanceof CopyException)
          || (t instanceof GemFireSecurityException) || (t instanceof ServerOperationException)
          || (t instanceof TransactionException) || (t instanceof CancelException)) {
        handleException(t, connection, retryCount, finalAttempt, timeoutFatal, cancelCriterion,
            endpointManager);
        return;
      } else if (throwable instanceof ServerOperationException) {
        title = null; // no message
        exceptionToThrow = (ServerOperationException) throwable;
        invalidateServer = false; // fix for bug #42225
      } else if (throwable instanceof FunctionException) {
        if (t instanceof InternalFunctionInvocationTargetException) {
          // Client server to re-execute for node failure
          handleException(t, connection, retryCount, finalAttempt, timeoutFatal, cancelCriterion,
              endpointManager);
          return;
        } else {
          title = null; // no message
          exceptionToThrow = (FunctionException) throwable;
        }
      } else if (throwable instanceof ServerConnectivityException
          && throwable.getMessage().equals("Connection error while authenticating user")) {
        title = null;
        if (logger.isDebugEnabled()) {
          logger.debug(throwable.getMessage(), throwable);
        }
      } else {
        title = throwable.toString();
        forceThrow = true;
      }
    }
    if (title != null) {
      connection.destroy();
      if (invalidateServer) {
        endpointManager.serverCrashed(connection.getEndpoint());
      }
      boolean logEnabled = warn ? logger.isWarnEnabled() : logger.isDebugEnabled();
      boolean msgNeeded = logEnabled || finalAttempt;
      if (msgNeeded) {
        final StringBuilder sb = getExceptionMessage(title, retryCount, finalAttempt, connection);
        final String msg = sb.toString();
        if (logEnabled) {
          if (warn) {
            logger.warn(msg);
          } else {
            logger.debug(msg, throwable);
          }
        }
        if (forceThrow || finalAttempt) {
          exceptionToThrow = new ServerConnectivityException(msg, cause);
        }
      }
    }
    if (exceptionToThrow != null) {
      throw exceptionToThrow;
    }
  }
This method will throw an exception if we need to stop the connection manager if there are
failures.
geode-core/src/main/java/org/apache/geode/cache/client/internal/OpExecutorImpl.java