specify connection type as direct or gateway on client #999

Author: ehennum

marklogic-client-api/src/main/java/com/marklogic/client/DatabaseClient.java

@@ -34,17 +34,19 @@
 import com.marklogic.client.pojo.PojoRepository;
 import com.marklogic.client.semantics.GraphManager;
 import com.marklogic.client.semantics.SPARQLQueryManager;
-import com.marklogic.client.DatabaseClientFactory.Authentication;
-import com.marklogic.client.DatabaseClientFactory.SSLHostnameVerifier;
 import com.marklogic.client.DatabaseClientFactory.SecurityContext;
 
-import javax.net.ssl.SSLContext;
-
 /**
  * A Database Client instantiates document and query managers and other objects
  * with shared access to a database.
  */
 public interface DatabaseClient {
+  /**
+   * Identifies whether the client connects directly to MarkLogic (the default) or
+   * by means of a gateway such as a load balancer.
+   */
+  enum ConnectionType {DIRECT, GATEWAY}
+
   /**
    * Starts a transaction.  You can pass the transaction to the read(), write(), or delete() methods
    * of a document manager or the search() method of a query manager to perform operations within a
@@ -100,22 +102,18 @@ public interface DatabaseClient {
   XMLDocumentManager newXMLDocumentManager();
 
   /**
-   * Creates a manager for long-running asynchronous write or query jobs
-   * with the default ConnectionPolicy.FOREST_HOSTS policy of creating
-   * a new connection for each host that has forests for the database.
+   * Creates a manager for long-running asynchronous write or query jobs.
+   * When the primary database client has the default ConnectionType.DIRECT
+   * connection type, the DataMovementManager creates a new connection
+   * for each host that has forests for the database.  When the primary
+   * database client has the ConnectionType.GATEWAY connection type
+   * (for instance, when connecting to a load balancer), the DataMovementManager
+   * uses the primary database client for all communication.
    * Don't forget to call dataMovementManager.release() when you're done with it.
    * @return	a manager supporting long-running asynchronous write or query jobs
    */
   DataMovementManager newDataMovementManager();
 
-  /**
-   * Creates a manager for long-running asynchronous write or query jobs.  Don't forget
-   * to call dataMovementManager.release() when you're done with it.
-   * @param policy	how the DataMovementManager should create connections
-   * @return	a manager supporting long-running asynchronous write or query jobs
-   */
-  DataMovementManager newDataMovementManager(ConnectionPolicy policy);
-
   /**
    * Creates a manager to query for database documents.
    * @return	a manager supporting search operations and lookup of values and tuples in indexes (also known as lexicons)
@@ -223,10 +221,10 @@ public interface DatabaseClient {
   ServerEvaluationCall newServerEval();
 
   /**
-   * Specifies whether to connect only using the host from the initial database connection
-   * or to create a new connection for each host that has forests for the database.
+   * How the client connects to MarkLogic.
+   * @return the connection type
    */
-  public enum ConnectionPolicy{PRIMARY_HOST, FOREST_HOSTS}
+  ConnectionType getConnectionType();
 
   String getHost();
 

marklogic-client-api/src/main/java/com/marklogic/client/DatabaseClientFactory.java

@@ -687,7 +687,25 @@ static public DatabaseClient newClient(String host, int port, String database) {
    * @return a new client for making database requests
    */
   static public DatabaseClient newClient(String host, int port, SecurityContext securityContext) {
-    return newClient(host, port, null, securityContext);
+    return newClient(host, port, null, securityContext, null);
+  }
+
+  /**
+   * Creates a client to access the database by means of a REST server.
+   *
+   * @param host the host with the REST server
+   * @param port the port for the REST server
+   * @param securityContext the security context created depending upon the
+   *            authentication type - BasicAuthContext, DigestAuthContext or KerberosAuthContext
+   *            and communication channel type (SSL)
+   * @param connectionType whether the client connects directly to the MarkLogic host
+   *            or using a gateway such as a load balancer
+   * @return a new client for making database requests
+   */
+  static public DatabaseClient newClient(String host, int port, SecurityContext securityContext,
+                                         DatabaseClient.ConnectionType connectionType)
+  {
+    return newClient(host, port, null, securityContext, connectionType);
   }
 
   /**
@@ -703,6 +721,27 @@ static public DatabaseClient newClient(String host, int port, SecurityContext se
    * @return a new client for making database requests
    */
   static public DatabaseClient newClient(String host, int port, String database, SecurityContext securityContext) {
+    return newClient(host, port, null, securityContext, null);
+  }
+
+  /**
+   * Creates a client to access the database by means of a REST server.
+   *
+   * @param host the host with the REST server
+   * @param port the port for the REST server
+   * @param database the database to access (default: configured database for
+   *            the REST server)
+   * @param securityContext the security context created depending upon the
+   *            authentication type - BasicAuthContext, DigestAuthContext or KerberosAuthContext
+   *            and communication channel type (SSL)
+   * @param connectionType whether the client connects directly to the MarkLogic host
+   *            or using a gateway such as a load balancer
+   * @return a new client for making database requests
+   */
+  static public DatabaseClient newClient(String host, int port, String database,
+                                         SecurityContext securityContext,
+                                         DatabaseClient.ConnectionType connectionType)
+  {
     String user = null;
     String password = null;
     Authentication type = null;
@@ -783,7 +822,9 @@ static public DatabaseClient newClient(String host, int port, String database, S
       }
     }
 
-    DatabaseClientImpl client = new DatabaseClientImpl(services, host, port, database, securityContext);
+    DatabaseClientImpl client = new DatabaseClientImpl(
+          services, host, port, database, securityContext, connectionType
+    );
     client.setHandleRegistry(getHandleRegistry().copy());
     return client;
   }
@@ -816,7 +857,7 @@ static private SecurityContext makeSecurityContext(String user, String password,
    */
   @Deprecated
   static public DatabaseClient newClient(String host, int port, String user, String password, Authentication type) {
-    return newClient(host, port, null, makeSecurityContext(user, password, type, null, null));
+    return newClient(host, port, null, makeSecurityContext(user, password, type, null, null), null);
   }
   /**
    * Creates a client to access the database by means of a REST server.
@@ -832,7 +873,7 @@ static public DatabaseClient newClient(String host, int port, String user, Strin
    */
   @Deprecated
   static public DatabaseClient newClient(String host, int port, String database, String user, String password, Authentication type) {
-    return newClient(host, port, database, makeSecurityContext(user, password, type, null, null));
+    return newClient(host, port, database, makeSecurityContext(user, password, type, null, null), null);
   }
   /**
    * Creates a client to access the database by means of a REST server.
@@ -848,7 +889,7 @@ static public DatabaseClient newClient(String host, int port, String database, S
    */
   @Deprecated
   static public DatabaseClient newClient(String host, int port, String user, String password, Authentication type, SSLContext context) {
-    return newClient(host, port, null, makeSecurityContext(user, password, type, context, SSLHostnameVerifier.COMMON));
+    return newClient(host, port, null, makeSecurityContext(user, password, type, context, SSLHostnameVerifier.COMMON), null);
   }
   /**
    * Creates a client to access the database by means of a REST server.
@@ -865,7 +906,7 @@ static public DatabaseClient newClient(String host, int port, String user, Strin
    */
   @Deprecated
   static public DatabaseClient newClient(String host, int port, String database, String user, String password, Authentication type, SSLContext context) {
-    return newClient(host, port, database, makeSecurityContext(user, password, type, context, SSLHostnameVerifier.COMMON));
+    return newClient(host, port, database, makeSecurityContext(user, password, type, context, SSLHostnameVerifier.COMMON), null);
   }
   /**
    * Creates a client to access the database by means of a REST server.
@@ -882,7 +923,7 @@ static public DatabaseClient newClient(String host, int port, String database, S
    */
   @Deprecated
   static public DatabaseClient newClient(String host, int port, String user, String password, Authentication type, SSLContext context, SSLHostnameVerifier verifier) {
-    return newClient(host, port, null, makeSecurityContext(user, password, type, context, verifier));
+    return newClient(host, port, null, makeSecurityContext(user, password, type, context, verifier), null);
   }
   /**
    * Creates a client to access the database by means of a REST server.
@@ -900,7 +941,7 @@ static public DatabaseClient newClient(String host, int port, String user, Strin
    */
   @Deprecated
   static public DatabaseClient newClient(String host, int port, String database, String user, String password, Authentication type, SSLContext context, SSLHostnameVerifier verifier) {
-    return newClient(host, port, database, makeSecurityContext(user, password, type, context, verifier));
+    return newClient(host, port, database, makeSecurityContext(user, password, type, context, verifier), null);
   }
 
   /**
@@ -978,6 +1019,7 @@ static public class Bean implements Serializable {
     private           Authentication        authentication;
     private           String                externalName;
     private           SecurityContext       securityContext;
+    private           DatabaseClient.ConnectionType connectionType;
     private           HandleFactoryRegistry handleRegistry =
       HandleFactoryRegistryImpl.newDefault();
 
@@ -1189,6 +1231,24 @@ public SecurityContext getSecurityContext() {
     public void setSecurityContext(SecurityContext securityContext) {
       this.securityContext = securityContext;
     }
+    /**
+     * Identifies whether the client connects directly with a MarkLogic host
+     * or by means of a gateway such as a load balancer.
+     * @return	the connection type
+     */
+    public DatabaseClient.ConnectionType getConnectionType() {
+      return connectionType;
+    }
+    /**
+     * Specify whether the client connects directly with a MarkLogic host
+     * or by means of a gateway such as a load balancer.
+     * @param connectionType	the connection type
+     */
+    public void setConnectionType(DatabaseClient.ConnectionType connectionType) {
+      this.connectionType = connectionType;
+    }
+
+
     /**
      * Returns the registry for associating
      * IO representation classes with handle factories.
@@ -1222,8 +1282,10 @@ public void registerDefaultHandles() {
      * @return	a new client for making database requests
      */
     public DatabaseClient newClient() {
-      DatabaseClientImpl client = (DatabaseClientImpl) DatabaseClientFactory.newClient(host, port, database,
-        makeSecurityContext(user, password, authentication, context, verifier));
+      DatabaseClientImpl client = (DatabaseClientImpl) DatabaseClientFactory.newClient(
+            host, port, database,
+            makeSecurityContext(user, password, authentication, context, verifier),
+            connectionType);
       client.setHandleRegistry(getHandleRegistry().copy());
       return client;
     }

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/DataMovementManager.java

@@ -16,12 +16,10 @@
 package com.marklogic.client.datamovement;
 
 import com.marklogic.client.DatabaseClient;
-import com.marklogic.client.query.QueryDefinition;
 import com.marklogic.client.query.StringQueryDefinition;
 import com.marklogic.client.query.StructuredQueryDefinition;
 import com.marklogic.client.query.RawCombinedQueryDefinition;
 import com.marklogic.client.query.RawStructuredQueryDefinition;
-import com.marklogic.client.datamovement.impl.DataMovementManagerImpl;
 
 import java.util.Iterator;
 
@@ -186,9 +184,10 @@ public interface DataMovementManager {
   public ForestConfiguration readForestConfig();
 
   /**
-   * Identify the DataMovementManager uses only the primary database connection
-   * or manages a connection for each host that has forests for the database.
-   * @return the connection policy
+   * Identify whether the DataMovementManager connects directly to each MarkLogic host
+   * with a forest for the database or whether the DataMovementManager uses a gateway
+   * such as a load balancer to communicate with the MarkLogic hosts.
+   * @return the connection type
    */
-  public DatabaseClient.ConnectionPolicy getConnectionPolicy();
+  public DatabaseClient.ConnectionType getConnectionType();
 }

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/HostAvailabilityListener.java

@@ -70,11 +70,6 @@ public class HostAvailabilityListener implements QueryFailureListener, WriteFail
   private ScheduledFuture<?> future;
   Set<QueryBatchListener> retryListenersSet = new HashSet<>();
   List<Class<?>> hostUnavailableExceptions = new ArrayList<>();
-  {
-    hostUnavailableExceptions.add(SocketException.class);
-    hostUnavailableExceptions.add(SSLException.class);
-    hostUnavailableExceptions.add(UnknownHostException.class);
-  }
 
   // Retry listener for Query batches, for which the list of URIs have been
   // retrieved from the server but the batch failed while applying the listener
@@ -109,6 +104,11 @@ public void processFailure(QueryBatch batch, Throwable throwable) {
   public HostAvailabilityListener(DataMovementManager moveMgr) {
     if (moveMgr == null) throw new IllegalArgumentException("moveMgr must not be null");
     this.moveMgr = moveMgr;
+    if (moveMgr.getConnectionType() == DatabaseClient.ConnectionType.DIRECT) {
+      hostUnavailableExceptions.add(SocketException.class);
+      hostUnavailableExceptions.add(SSLException.class);
+      hostUnavailableExceptions.add(UnknownHostException.class);
+    }
   }
 
   /** If a host becomes unavailable (SocketException, SSLException,
@@ -132,7 +132,7 @@ public HostAvailabilityListener withSuspendTimeForHostUnavailable(Duration
    * @return this instance (for method chaining)
    */
   public HostAvailabilityListener withMinHosts(int numHosts) {
-    if (moveMgr.getConnectionPolicy() == DatabaseClient.ConnectionPolicy.PRIMARY_HOST) {
+    if (moveMgr.getConnectionType() == DatabaseClient.ConnectionType.GATEWAY) {
       if (numHosts != 1) {
         throw new IllegalArgumentException("numHosts must be 1 when using only the primary host for the connection");
       }
@@ -224,22 +224,20 @@ public void processFailure(QueryBatchException queryBatch) {
   }
 
   private synchronized boolean processException(Batcher batcher, Throwable throwable, String host) {
-    // we only do something if this throwable is on our list of exceptions
-    // which we consider marking a host as unavilable
-    boolean isHostUnavailableException = isHostUnavailableException(throwable, new HashSet<>());
-    return (moveMgr.getConnectionPolicy() == DatabaseClient.ConnectionPolicy.PRIMARY_HOST) ?
-           processPrimaryHostException(batcher, throwable, host, isHostUnavailableException) :
-           processForestHostException(batcher, throwable, host, isHostUnavailableException);
+    return (moveMgr.getConnectionType() == DatabaseClient.ConnectionType.GATEWAY) ?
+           processGatewayException(batcher, throwable, host) :
+           processForestHostException(batcher, throwable, host);
   }
 
-  private boolean processPrimaryHostException(Batcher batcher, Throwable throwable, String host, boolean isHostUnavailableException) {
-    // TODO:  currently, cancelling the job because the load balancer is down would require
-    // a custom listener; determine whether the approach could be simplified with an array
-    // of exceptions different from the exceptions that indicate the MarkLogic host is unavailable
-    return isHostUnavailableException;
+  private boolean processGatewayException(Batcher batcher, Throwable throwable, String host) {
+    // if the nested retry failed, assume the MarkLogic cluster is unavailable
+    return false;
   }
 
-  private boolean processForestHostException(Batcher batcher, Throwable throwable, String host, boolean isHostUnavailableException) {
+  private boolean processForestHostException(Batcher batcher, Throwable throwable, String host) {
+    // we only do something if this throwable is on our list of exceptions
+    // which we consider marking a host as unavilable
+    boolean isHostUnavailableException = isHostUnavailableException(throwable, new HashSet<>());
     boolean shouldWeRetry = isHostUnavailableException;
     if ( isHostUnavailableException == true ) {
       ForestConfiguration existingForestConfig = batcher.getForestConfig();

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/DataMovementManagerImpl.java

@@ -30,22 +30,15 @@
 import com.marklogic.client.datamovement.JobTicket;
 import com.marklogic.client.datamovement.NoResponseListener;
 import com.marklogic.client.datamovement.QueryBatcher;
-import com.marklogic.client.datamovement.impl.QueryJobReportListener;
 import com.marklogic.client.datamovement.WriteBatcher;
 import com.marklogic.client.datamovement.JobReport;
-import com.marklogic.client.datamovement.impl.WriteJobReportListener;
-import com.marklogic.client.datamovement.impl.QueryBatcherImpl;
-import com.marklogic.client.datamovement.impl.DataMovementServices;
-import com.marklogic.client.datamovement.impl.WriteBatcherImpl;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
-import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class DataMovementManagerImpl implements DataMovementManager {
@@ -54,17 +47,19 @@ public class DataMovementManagerImpl implements DataMovementManager {
   private static ConcurrentHashMap<String, JobTicket> activeJobs = new ConcurrentHashMap<>();
   private ForestConfiguration forestConfig;
   private DatabaseClient primaryClient;
-  private DatabaseClient.ConnectionPolicy connectPolicy;
+  private DatabaseClient.ConnectionType connectionType;
   // clientMap key is the hostname_database
   private Map<String,DatabaseClient> clientMap = new HashMap<>();
 
-  public DataMovementManagerImpl(DatabaseClient client, DatabaseClient.ConnectionPolicy connectPolicy) {
-    setPrimaryClient(client);
-    clientMap.put(primaryClient.getHost(), primaryClient);
-    this.connectPolicy = connectPolicy;
-    if (connectPolicy == DatabaseClient.ConnectionPolicy.PRIMARY_HOST) {
+  public DataMovementManagerImpl(DatabaseClient client) {
+    connectionType = client.getConnectionType();
+    if (connectionType == DatabaseClient.ConnectionType.GATEWAY) {
       forestConfig = new AnyForestConfiguration(client);
     }
+
+    setPrimaryClient(client);
+
+    clientMap.put(primaryClient.getHost(), primaryClient);
   }
 
   @Override
@@ -168,7 +163,7 @@ private ForestConfiguration getForestConfig() {
 
   @Override
   public ForestConfiguration readForestConfig() {
-    if (connectPolicy == DatabaseClient.ConnectionPolicy.FOREST_HOSTS) {
+    if (getConnectionType() == DatabaseClient.ConnectionType.DIRECT) {
      forestConfig = service.readForestConfig();
     }
     return forestConfig;
@@ -208,8 +203,8 @@ public JobTicket getActiveJob(String jobId) {
   }
 
   @Override
-  public DatabaseClient.ConnectionPolicy getConnectionPolicy() {
-    return connectPolicy;
+  public DatabaseClient.ConnectionType getConnectionType() {
+    return connectionType;
   }
 
   public DataMovementServices getDataMovementServices() {