/*
 * 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.cache.client.internal;

import java.io.IOException;
import java.io.NotSerializableException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.BufferUnderflowException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.CopyException;
import org.apache.geode.GemFireException;
import org.apache.geode.GemFireIOException;
import org.apache.geode.SerializationException;
import org.apache.geode.cache.CacheRuntimeException;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.SynchronizationCommitConflictException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.client.NoAvailableServersException;
import org.apache.geode.cache.client.ServerConnectivityException;
import org.apache.geode.cache.client.ServerOperationException;
import org.apache.geode.cache.client.ServerRefusedConnectionException;
import org.apache.geode.cache.client.SubscriptionNotEnabledException;
import org.apache.geode.cache.client.internal.ExecuteFunctionOp.ExecuteFunctionOpImpl;
import org.apache.geode.cache.client.internal.ExecuteRegionFunctionOp.ExecuteRegionFunctionOpImpl;
import org.apache.geode.cache.client.internal.QueueManager.QueueConnections;
import org.apache.geode.cache.client.internal.pooling.ConnectionDestroyedException;
import org.apache.geode.cache.client.internal.pooling.ConnectionManager;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionInvocationTargetException;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.internal.cache.PoolManagerImpl;
import org.apache.geode.internal.cache.PutAllPartialResultException;
import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException;
import org.apache.geode.internal.cache.tier.BatchException;
import org.apache.geode.internal.cache.tier.sockets.MessageTooLargeException;
import org.apache.geode.internal.cache.wan.BatchException70;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.security.AuthenticationRequiredException;
import org.apache.geode.security.GemFireSecurityException;

/**
 * Called from the client and execute client to server requests against servers. Handles retrying to
 * different servers, and marking servers dead if we get exception from them.
 *
 * @since GemFire 5.7
 */
public class OpExecutorImpl implements ExecutablePool {
  private static final Logger logger = LogService.getLogger();

  private static final boolean TRY_SERVERS_ONCE =
      Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "PoolImpl.TRY_SERVERS_ONCE");
  static final int TX_RETRY_ATTEMPT =
      Integer.getInteger(DistributionConfig.GEMFIRE_PREFIX + "txRetryAttempt", 500);

  private final ConnectionManager connectionManager;
  private final int retryAttempts;
  private final long serverTimeout;
  private final EndpointManager endpointManager;
  private final RegisterInterestTracker riTracker;
  private final QueueManager queueManager;
  private final CancelCriterion cancelCriterion;
  private /* final */ PoolImpl pool;
  private final ThreadLocal<Boolean> serverAffinity = ThreadLocal.withInitial(() -> Boolean.FALSE);
  private boolean serverAffinityFailover = false;
  private final ThreadLocal<ServerLocation> affinityServerLocation = new ThreadLocal<>();

  private final ThreadLocal<Integer> affinityRetryCount = ThreadLocal.withInitial(() -> 0);

  public OpExecutorImpl(ConnectionManager connectionManager, QueueManager queueManager,
      EndpointManager endpointManager, RegisterInterestTracker riTracker, int retryAttempts,
      long serverTimeout, CancelCriterion cancelCriterion,
      PoolImpl pool) {
    this.connectionManager = connectionManager;
    this.queueManager = queueManager;
    this.endpointManager = endpointManager;
    this.riTracker = riTracker;
    this.retryAttempts = retryAttempts;
    this.serverTimeout = serverTimeout;
    this.cancelCriterion = cancelCriterion;
    this.pool = pool;
  }

  @Override
  public Object execute(Op op) {
    return execute(op, retryAttempts);
  }

  @Override
  public Object execute(Op op, int retries) {
    if (serverAffinity.get()) {
      ServerLocation loc = affinityServerLocation.get();
      if (loc == null) {
        loc = getNextOpServerLocation();
        affinityServerLocation.set(loc);
        if (logger.isDebugEnabled()) {
          logger.debug("setting server affinity to {}", affinityServerLocation.get());
        }
      }
      return executeWithServerAffinity(loc, op);
    }

    Connection conn = connectionManager.borrowConnection(serverTimeout);
    try {
      Set<ServerLocation> attemptedServers = null;

      for (int attempt = 0; true; attempt++) {
        // when an op is retried we may need to try to recover the previous
        // attempt's version stamp
        if (attempt == 1 && (op instanceof AbstractOp)) {
          AbstractOp absOp = (AbstractOp) op;
          absOp.getMessage().setIsRetry();
        }
        try {
          authenticateIfRequired(conn, op);
          return executeWithPossibleReAuthentication(conn, op);
        } catch (MessageTooLargeException e) {
          throw new GemFireIOException("unable to transmit message to server", e);
        } catch (Exception e) {
          handleException(e, conn, attempt, attempt >= retries && retries != -1);
          if (null == attemptedServers) {
            // don't allocate this until we need it
            attemptedServers = new HashSet<>();
          }
          attemptedServers.add(conn.getServer());
          try {
            conn = connectionManager.exchangeConnection(conn, attemptedServers);
          } catch (NoAvailableServersException nse) {
            // if retries is -1, don't try again after the last server has failed
            if (retries == -1 || TRY_SERVERS_ONCE) {
              handleException(e, conn, attempt, true);
            } else {
              // try one of the failed servers again, until we exceed the retry attempts.
              attemptedServers.clear();
              try {
                conn = connectionManager.exchangeConnection(conn, attemptedServers);
              } catch (NoAvailableServersException nse2) {
                handleException(e, conn, attempt, true);
              }
            }
          }
        }
      }
    } finally {
      connectionManager.returnConnection(conn);
    }
  }

  /**
   * execute the given op on the given server. If the server cannot be reached, sends a
   * TXFailoverOp, then retries the given op
   *
   * @param loc the server to execute the op on
   * @param op the op to execute
   * @return the result of execution
   */
  Object executeWithServerAffinity(ServerLocation loc, Op op) {
    final int initialRetryCount = getAffinityRetryCount();
    try {
      try {
        return executeOnServer(loc, op, true, false);
      } catch (ServerConnectivityException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("caught exception while executing with affinity:{}", e.getMessage(), e);
        }
        if (!serverAffinityFailover || e instanceof ServerOperationException) {
          throw e;
        }
        int retryCount = getAffinityRetryCount();
        if ((retryAttempts != -1 && retryCount >= retryAttempts)
            || retryCount > TX_RETRY_ATTEMPT) {
          // prevent stack overflow
          throw e;
        }
        setAffinityRetryCount(retryCount + 1);
      }
      affinityServerLocation.set(null);
      if (logger.isDebugEnabled()) {
        logger.debug("reset server affinity: attempting txFailover");
      }
      // send TXFailoverOp, so that new server can
      // do bootstrapping, then re-execute original op
      AbstractOp absOp = (AbstractOp) op;
      absOp.getMessage().setIsRetry();
      int transactionId = absOp.getMessage().getTransactionId();
      // for CommitOp we do not have transactionId in AbstractOp
      // so set it explicitly for TXFailoverOp
      TXFailoverOp.execute(pool, transactionId);

      if (op instanceof ExecuteRegionFunctionOpImpl) {
        op = new ExecuteRegionFunctionOpImpl((ExecuteRegionFunctionOpImpl) op, (byte) 1,
            new HashSet<>());
        ((ExecuteRegionFunctionOpImpl) op).getMessage().setTransactionId(transactionId);
      } else if (op instanceof ExecuteFunctionOpImpl) {
        op = new ExecuteFunctionOpImpl((ExecuteFunctionOpImpl) op, (byte) 1/* isReExecute */);
        ((ExecuteFunctionOpImpl) op).getMessage().setTransactionId(transactionId);
      }
      return pool.execute(op);
    } finally {
      if (initialRetryCount == 0) {
        setAffinityRetryCount(0);
      }
    }
  }

  @Override
  public void setupServerAffinity(boolean allowFailover) {
    if (logger.isDebugEnabled()) {
      logger.debug("setting up server affinity");
    }
    serverAffinityFailover = allowFailover;
    serverAffinity.set(Boolean.TRUE);
  }

  @Override
  public void releaseServerAffinity() {
    if (logger.isDebugEnabled()) {
      logger.debug("reset server affinity");
    }
    serverAffinity.set(Boolean.FALSE);
    affinityServerLocation.set(null);
  }

  @Override
  public ServerLocation getServerAffinityLocation() {
    return affinityServerLocation.get();
  }

  int getAffinityRetryCount() {
    return affinityRetryCount.get();
  }

  void setAffinityRetryCount(int retryCount) {
    affinityRetryCount.set(retryCount);
  }

  @Override
  public void setServerAffinityLocation(ServerLocation serverLocation) {
    assert affinityServerLocation.get() == null;
    affinityServerLocation.set(serverLocation);
  }

  public ServerLocation getNextOpServerLocation() {
    Connection conn = connectionManager.borrowConnection(serverTimeout);
    try {
      return conn.getServer();
    } finally {
      connectionManager.returnConnection(conn);
    }
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.cache.client.internal.OpExecutor#executeOn(org.apache.geode.distributed.
   * internal.ServerLocation, org.apache.geode.cache.client.internal.Op)
   */
  @Override
  public Object executeOn(ServerLocation server, Op op) {
    return executeOn(server, op, true, false);
  }

  @Override
  public Object executeOn(ServerLocation p_server, Op op, boolean accessed,
      boolean onlyUseExistingCnx) {
    ServerLocation server = p_server;
    if (serverAffinity.get()) {
      ServerLocation affinityserver = affinityServerLocation.get();
      if (affinityserver != null) {
        server = affinityserver;
      } else {
        affinityServerLocation.set(server);
      }
      // redirect to executeWithServerAffinity so that we
      // can send a TXFailoverOp.
      return executeWithServerAffinity(server, op);
    }
    return executeOnServer(server, op, accessed, onlyUseExistingCnx);
  }

  protected Object executeOnServer(ServerLocation p_server, Op op, boolean accessed,
      boolean onlyUseExistingCnx) {
    boolean returnCnx = true;
    boolean pingOp = (op instanceof PingOp.PingOpImpl);
    Connection conn = null;
    if (pingOp) {
      // currently for pings we prefer to queue clientToServer cnx so that we will
      // not create a pooled cnx when all we have is queue cnxs.
      if (queueManager != null) {
        // see if our QueueManager has a connection to this server that we can send
        // the ping on.
        Endpoint ep = endpointManager.getEndpointMap().get(p_server);
        if (ep != null) {
          QueueConnections qcs = queueManager.getAllConnectionsNoWait();
          conn = qcs.getConnection(ep);
          if (conn != null) {
            // we found one to do the ping on
            returnCnx = false;
          }
        }
      }
    }
    if (conn == null) {
      conn = connectionManager.borrowConnection(p_server, onlyUseExistingCnx);
    }
    try {
      return executeWithPossibleReAuthentication(conn, op);
    } catch (Exception e) {
      handleException(e, conn, 0, true);
      // this shouldn't actually be reached, handle exception will throw something
      throw new ServerConnectivityException("Received error connecting to server", e);
    } finally {
      if (serverAffinity.get() && affinityServerLocation.get() == null) {
        if (logger.isDebugEnabled()) {
          logger.debug("setting server affinity to {} server:{}", conn.getEndpoint().getMemberId(),
              conn.getServer());
        }
        affinityServerLocation.set(conn.getServer());
      }
      if (returnCnx) {
        connectionManager.returnConnection(conn, accessed);
      }
    }
  }

  /*
   * (non-Javadoc)
   *
   * @see
   * org.apache.geode.cache.client.internal.ExecutablePool#executeOnPrimary(org.apache.geode.cache.
   * client.internal.Op)
   */
  @Override
  public Object executeOnPrimary(Op op) {
    if (queueManager == null) {
      throw new SubscriptionNotEnabledException();
    }

    HashSet<ServerLocation> attemptedPrimaries = new HashSet<>();
    while (true) {
      Connection primary = queueManager.getAllConnections().getPrimary();
      try {
        return executeWithPossibleReAuthentication(primary, op);
      } catch (Exception e) {
        boolean finalAttempt = !attemptedPrimaries.add(primary.getServer());
        handleException(e, primary, 0, finalAttempt);
        // we shouldn't reach this code, but just in case
        if (finalAttempt) {
          throw new ServerConnectivityException("Tried the same primary server twice.", e);
        }
      }
    }
  }

  @Override
  public void executeOnAllQueueServers(Op op) {
    if (queueManager == null) {
      throw new SubscriptionNotEnabledException();
    }

    RuntimeException lastException = null;

    QueueConnections connections = queueManager.getAllConnectionsNoWait();
    Connection primary = connections.getPrimary();
    if (primary != null) {
      try {
        executeWithPossibleReAuthentication(primary, op);
      } catch (Exception e) {
        try {
          handleException(e, primary, 0, false);
        } catch (RuntimeException e2) {
          lastException = e2;
        }
      }
    }

    List<Connection> backups = connections.getBackups();
    for (Connection conn : backups) {
      try {
        executeWithPossibleReAuthentication(conn, op);
      } catch (Exception e) {
        try {
          handleException(e, conn, 0, false);
        } catch (RuntimeException e2) {
          lastException = e2;
        }
      }
    }

    if (lastException != null) {
      throw lastException;
    }
  }

  /*
   * (non-Javadoc)
   *
   * @see
   * org.apache.geode.cache.client.internal.ExecutablePool#executeOnAllQueueServers(org.apache.geode
   * .cache.client.internal.Op)
   */
  @Override
  public Object executeOnQueuesAndReturnPrimaryResult(Op op) {
    if (queueManager == null) {
      throw new SubscriptionNotEnabledException();
    }
    QueueConnections connections = queueManager.getAllConnections();

    List backups = connections.getBackups();
    if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
      logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "sending {} to backups: {}", op, backups);
    }
    for (int i = backups.size() - 1; i >= 0; i--) {
      Connection conn = (Connection) backups.get(i);
      try {
        executeWithPossibleReAuthentication(conn, op);
      } catch (Exception e) {
        handleException(e, conn, 0, false);
      }
    }

    Connection primary = connections.getPrimary();
    HashSet<ServerLocation> attemptedPrimaries = new HashSet<>();
    while (true) {
      try {
        if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
          logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "sending {} to primary: {}", op, primary);
        }
        return executeWithPossibleReAuthentication(primary, op);
      } catch (Exception e) {
        if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
          logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "caught exception sending to primary {}",
              e.getMessage(), e);
        }
        boolean finalAttempt = !attemptedPrimaries.add(primary.getServer());
        handleException(e, primary, 0, finalAttempt);
        primary = queueManager.getAllConnections().getPrimary();
        // we shouldn't reach this code, but just in case
        if (finalAttempt) {
          throw new ServerConnectivityException("Tried the same primary server twice.", e);
        }
      }
    }
  }

  /**
   * Used by GatewayBatchOp
   */
  @Override
  public Object executeOn(Connection conn, Op op, boolean timeoutFatal) {
    try {
      return executeWithPossibleReAuthentication(conn, op);
    } catch (Exception e) {
      handleException(op, e, conn, 0, true, timeoutFatal);
      // this shouldn't actually be reached, handle exception will throw something
      throw new ServerConnectivityException("Received error connecting to server", e);
    }
  }

  /**
   * This is used by unit tests
   */
  @Override
  public Object executeOn(Connection conn, Op op) {
    return executeOn(conn, op, false);
  }

  @Override
  public RegisterInterestTracker getRITracker() {
    return riTracker;
  }

  /**
   * This method will throw an exception if we need to stop the connection manager if there are
   * failures.
   */
  protected void handleException(Throwable e, Connection conn, int retryCount,
      boolean finalAttempt) {
    handleException(e, conn, retryCount, finalAttempt, false/* timeoutFatal */);
  }

  protected void handleException(Op op, Throwable e, Connection conn, int retryCount,
      boolean finalAttempt, boolean timeoutFatal) throws CacheRuntimeException {
    if (op instanceof AuthenticateUserOp.AuthenticateUserOpImpl) {
      if (e instanceof GemFireSecurityException) {
        throw (GemFireSecurityException) e;
      } else if (e instanceof ServerRefusedConnectionException) {
        throw (ServerRefusedConnectionException) e;
      }
    }
    handleException(e, conn, retryCount, finalAttempt, timeoutFatal);
  }

  protected void handleException(Throwable e, Connection conn, int retryCount, boolean finalAttempt,
      boolean timeoutFatal) throws CacheRuntimeException {
    GemFireException exToThrow = null;
    String title;
    boolean invalidateServer = true;
    boolean warn = true;
    boolean forceThrow = false;
    Throwable cause = e;

    cancelCriterion.checkCancelInProgress(e);

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

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

    if (e instanceof MessageTooLargeException) {
      title = null;
      exToThrow = new GemFireIOException("message is too large to transmit", e);
    } else if (e instanceof NotSerializableException) {
      title = null; // no message
      exToThrow = new SerializationException("Pool message failure", e);
    } else if (e instanceof BatchException || e instanceof BatchException70) {
      title = null; // no message
      exToThrow = new ServerOperationException(e);
    } else if (e instanceof RegionDestroyedException) {
      invalidateServer = false;
      title = null;
      exToThrow = (RegionDestroyedException) e;
    } else if (e instanceof GemFireSecurityException) {
      title = null;
      exToThrow = new ServerOperationException(e);
    } else if (e instanceof SerializationException) {
      title = null; // no message
      exToThrow = new ServerOperationException(e);
    } else if (e instanceof CopyException) {
      title = null; // no message
      exToThrow = new ServerOperationException(e);
    } else if (e instanceof ClassNotFoundException) {
      title = null; // no message
      exToThrow = new ServerOperationException(e);
    } else if (e instanceof TransactionException) {
      title = null; // no message
      exToThrow = (TransactionException) e;
      invalidateServer = false;
    } else if (e instanceof SynchronizationCommitConflictException) {
      title = null;
      exToThrow = (SynchronizationCommitConflictException) e;
      invalidateServer = false;
    } else if (e instanceof SocketException) {
      if ("Socket closed".equals(e.getMessage()) || "Connection reset".equals(e.getMessage())
          || "Connection refused: connect".equals(e.getMessage())
          || "Connection refused".equals(e.getMessage())) {
        title = e.getMessage();
      } else {
        title = "SocketException";
      }
    } else if (e instanceof SocketTimeoutException) {
      invalidateServer = timeoutFatal;
      title = "socket timed out on client";
      cause = null;
    } else if (e instanceof ConnectionDestroyedException) {
      invalidateServer = false;
      title = "connection was asynchronously destroyed";
      cause = null;
    } else if (e instanceof java.io.EOFException) {
      title = "closed socket on server";
    } else if (e instanceof IOException) {
      title = "IOException";
    } else if (e instanceof BufferUnderflowException) {
      title = "buffer underflow reading from server";
    } else if (e instanceof CancelException) {
      title = "Cancelled";
      warn = false;
    } else if (e instanceof InternalFunctionInvocationTargetException) {
      // In this case, function will be re executed
      title = null;
      exToThrow = (InternalFunctionInvocationTargetException) e;
    } else if (e instanceof FunctionInvocationTargetException) {
      // in this case function will not be re executed
      title = null;
      exToThrow = (GemFireException) e;
    } else if (e instanceof PutAllPartialResultException) {
      title = null;
      exToThrow = (PutAllPartialResultException) e;
      invalidateServer = false;
    } else {
      Throwable t = e.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, conn, retryCount, finalAttempt, timeoutFatal);
        return;
      } else if (e instanceof ServerOperationException) {
        title = null; // no message
        exToThrow = (ServerOperationException) e;
        invalidateServer = false; // fix for bug #42225
      } else if (e instanceof FunctionException) {
        if (t instanceof InternalFunctionInvocationTargetException) {
          // Client server to re-execute for node failure
          handleException(t, conn, retryCount, finalAttempt, timeoutFatal);
          return;
        } else {
          title = null; // no message
          exToThrow = (FunctionException) e;
        }
      } else if (e instanceof ServerConnectivityException
          && e.getMessage().equals("Connection error while authenticating user")) {
        title = null;
        if (logger.isDebugEnabled()) {
          logger.debug(e.getMessage(), e);
        }
      } else {
        title = e.toString();
        forceThrow = true;
      }
    }
    if (title != null) {
      conn.destroy();
      if (invalidateServer) {
        endpointManager.serverCrashed(conn.getEndpoint());
      }
      boolean logEnabled = warn ? logger.isWarnEnabled() : logger.isDebugEnabled();
      boolean msgNeeded = logEnabled || finalAttempt;
      if (msgNeeded) {
        final StringBuffer sb = getExceptionMessage(title, retryCount, finalAttempt, conn);
        final String msg = sb.toString();
        if (logEnabled) {
          if (warn) {
            logger.warn(msg);
          } else {
            logger.debug(msg, e);
          }
        }
        if (forceThrow || finalAttempt) {
          exToThrow = new ServerConnectivityException(msg, cause);
        }
      }
    }
    if (exToThrow != null) {
      throw exToThrow;
    }
  }

  private StringBuffer getExceptionMessage(String exceptionName, int retryCount,
      boolean finalAttempt, Connection connection) {
    StringBuffer message = new StringBuffer(200);
    message.append("Pool unexpected ").append(exceptionName);
    if (connection != null) {
      message.append(" connection=").append(connection);
    }
    if (retryCount > 0) {
      message.append(" attempt=").append(retryCount + 1);
    }
    message.append(')');
    if (finalAttempt) {
      message.append(". Server unreachable: could not connect after ").append(retryCount + 1)
          .append(" attempts");
    }
    return message;
  }

  private void authenticateIfRequired(Connection conn, Op op) {
    if (!conn.getServer().getRequiresCredentials()) {
      return;
    }

    if (pool == null) {
      PoolImpl poolImpl =
          (PoolImpl) PoolManagerImpl.getPMI().find(endpointManager.getPoolName());
      if (poolImpl == null) {
        return;
      }
      pool = poolImpl;
    }
    if (pool.getMultiuserAuthentication()) {
      if (((AbstractOp) op).needsUserId()) {
        UserAttributes ua = UserAttributes.userAttributes.get();
        if (ua != null) {
          if (!ua.getServerToId().containsKey(conn.getServer())) {
            authenticateMultiuser(pool, conn, ua);
          }
        }
      }
    } else if (((AbstractOp) op).needsUserId()) {
      // This should not be reached, but keeping this code here in case it is
      // reached.
      if (conn.getServer().getUserId() == -1) {
        Connection connImpl = conn.getWrappedConnection();
        conn.getServer().setUserId((Long) AuthenticateUserOp.executeOn(connImpl, pool));
        if (logger.isDebugEnabled()) {
          logger.debug(
              "OpExecutorImpl.execute() - single user mode - authenticated this user on {}", conn);
        }
      }
    }
  }

  private void authenticateMultiuser(PoolImpl pool, Connection conn, UserAttributes ua) {
    try {
      Long userId =
          (Long) AuthenticateUserOp.executeOn(conn.getServer(), pool, ua.getCredentials());
      if (userId != null) {
        ua.setServerToId(conn.getServer(), userId);
        if (logger.isDebugEnabled()) {
          logger.debug("OpExecutorImpl.execute() - multiuser mode - authenticated this user on {}",
              conn);
        }
      }
    } catch (ServerConnectivityException sce) {
      Throwable cause = sce.getCause();
      if (cause instanceof IOException || cause instanceof BufferUnderflowException
          || cause instanceof CancelException
          || (sce.getMessage() != null
              && (sce.getMessage().contains("Could not create a new connection to server")
                  || sce.getMessage().contains("socket timed out on client")
                  || sce.getMessage().contains("connection was asynchronously destroyed")))) {
        throw new ServerConnectivityException("Connection error while authenticating user");
      } else {
        throw sce;
      }
    }
  }

  private Object executeWithPossibleReAuthentication(Connection conn, Op op) throws Exception {
    try {
      return conn.execute(op);

    } catch (ServerConnectivityException sce) {
      Throwable cause = sce.getCause();
      if ((cause instanceof AuthenticationRequiredException
          && "User authorization attributes not found.".equals(cause.getMessage()))
          || sce.getMessage().contains("Connection error while authenticating user")) {
        // (ashetkar) Need a cleaner way of doing above check.
        // 2nd exception-message above is from AbstractOp.sendMessage()

        PoolImpl pool =
            (PoolImpl) PoolManagerImpl.getPMI().find(endpointManager.getPoolName());
        if (!pool.getMultiuserAuthentication()) {
          Connection connImpl = conn.getWrappedConnection();
          conn.getServer().setUserId((Long) AuthenticateUserOp.executeOn(connImpl, this));
          return conn.execute(op);
        } else {
          UserAttributes ua = UserAttributes.userAttributes.get();
          if (ua != null) {
            authenticateMultiuser(pool, conn, ua);
          }
          return conn.execute(op);
        }

      } else {
        throw sce;
      }
    }
  }

}