#744 - WriteBatcher hangs; #668 - White list and black list; #666 - Null host names; #579 - Job not stopped

Author: vivekmuniyandi

src/main/java/com/marklogic/client/datamovement/FilteredForestConfiguration.java

@@ -15,11 +15,18 @@
  */
 package com.marklogic.client.datamovement;
 
+import com.marklogic.client.datamovement.Forest.HostType;
+import com.marklogic.client.datamovement.impl.ForestConfigurationImpl;
 import com.marklogic.client.datamovement.impl.ForestImpl;
+
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
@@ -71,6 +78,11 @@ public Forest[] listForests() {
       if ( renames.containsKey(openReplicaHost) ) {
         openReplicaHost = renames.get(openReplicaHost);
       }
+      String requestHost = forest.getRequestHost();
+      if ( requestHost != null ) requestHost = requestHost.toLowerCase();
+      if ( renames.containsKey(requestHost) ) {
+        requestHost = renames.get(requestHost);
+      }
       String alternateHost = forest.getAlternateHost();
       if ( alternateHost != null ) alternateHost = alternateHost.toLowerCase();
       if ( renames.containsKey(alternateHost) ) {
@@ -81,12 +93,19 @@ public Forest[] listForests() {
       if ( renames.containsKey(host) ) {
         host = renames.get(host);
       }
-      return new ForestImpl(host, openReplicaHost, alternateHost, forest.getDatabaseName(),
+      return new ForestImpl(host, openReplicaHost, requestHost, alternateHost, forest.getDatabaseName(),
         forest.getForestName(), forest.getForestId(), forest.isUpdateable(), false);
     }).toArray(Forest[]::new);
 
-    String[] validHosts = Stream.of(renamedForests).flatMap( forest ->
-      Stream.of(new String[] {forest.getHost(), forest.getOpenReplicaHost(), forest.getAlternateHost()})
+    String[] validHosts = Stream.of(renamedForests).flatMap( forest -> {
+      String[] hostArray;
+      if((blackList.contains(forest.getPreferredHost()) || blackList.contains(forest.getHost()))
+          && forest.getPreferredHostType() == HostType.REQUEST_HOST) {
+        hostArray = new String[] {forest.getOpenReplicaHost(), forest.getAlternateHost()};
+      } else {
+        hostArray = new String[] {forest.getHost(), forest.getOpenReplicaHost(), forest.getAlternateHost(), forest.getRequestHost()};
+      }
+      return Stream.of(hostArray)
         .filter( host -> host != null )
         .map   ( host -> host.toLowerCase() )
         .filter( host -> {
@@ -100,13 +119,14 @@ public Forest[] listForests() {
           }
           // this host is valid, add it to the list
           return true;
-        })
-    ).distinct().toArray(String[]::new);
+        });
+    }).distinct().toArray(String[]::new);
 
     Stream<Forest> replaced = Stream.of(renamedForests);
     if ( blackList.size() > 0 ) {
       replaced = replaced.map( forest -> {
-        if ( blackList.contains(forest.getPreferredHost()) ) {
+        if ( blackList.contains(forest.getPreferredHost()) ||
+            (forest.getPreferredHostType() == HostType.REQUEST_HOST && blackList.contains(forest.getHost()))) {
           if ( validHosts.length == 0 ) {
             throw new IllegalStateException("White list or black list rules are too restrictive:" +
               " no valid hosts are left");
@@ -117,15 +137,21 @@ public Forest[] listForests() {
           if ( blackList.contains(openReplicaHost) ) {
             openReplicaHost = replaceHost(openReplicaHost, validHosts);
           }
+          String requestHost = forest.getRequestHost();
+          String host = forest.getHost();
+          if ( blackList.contains(requestHost) ) {
+            requestHost = replaceHost(requestHost, validHosts);
+            host = requestHost;
+          }
           String alternateHost = forest.getAlternateHost();
           if ( blackList.contains(alternateHost) ) {
             alternateHost = replaceHost(alternateHost, validHosts);
           }
-          String host = forest.getHost();
           if ( blackList.contains(host) ) {
             host = replaceHost(host, validHosts);
+            if(requestHost != null) requestHost = host;
           }
-          return new ForestImpl(host, openReplicaHost, alternateHost, forest.getDatabaseName(),
+          return new ForestImpl(host, openReplicaHost, requestHost, alternateHost, forest.getDatabaseName(),
             forest.getForestName(), forest.getForestId(), forest.isUpdateable(), false);
         } else {
           return forest;
@@ -145,6 +171,10 @@ public Forest[] listForests() {
           if ( ! whiteList.contains(openReplicaHost) ) {
             openReplicaHost = replaceHost(openReplicaHost, validHosts);
           }
+          String requestHost = forest.getRequestHost();
+          if ( ! whiteList.contains(requestHost) ) {
+            requestHost = replaceHost(requestHost, validHosts);
+          }
           String alternateHost = forest.getAlternateHost();
           if ( ! whiteList.contains(alternateHost) ) {
             alternateHost = replaceHost(alternateHost, validHosts);
@@ -153,7 +183,7 @@ public Forest[] listForests() {
           if ( ! whiteList.contains(host) ) {
             host = replaceHost(host, validHosts);
           }
-          return new ForestImpl(host, openReplicaHost, alternateHost, forest.getDatabaseName(),
+          return new ForestImpl(host, openReplicaHost, requestHost, alternateHost, forest.getDatabaseName(),
             forest.getForestName(), forest.getForestId(), forest.isUpdateable(), false);
         } else {
           return forest;
@@ -182,6 +212,7 @@ private String replaceHost(String invalidHost, String[] validHosts) {
    */
   public FilteredForestConfiguration withBlackList(String... hostNames) {
     if ( whiteList.size() > 0 ) throw new IllegalStateException("whiteList already initialized");
+    if ( hostNames == null ) throw new IllegalArgumentException("hostNames must not be null");
     for ( String hostName : hostNames ) {
       if ( hostName != null ) blackList.add(hostName.toLowerCase());
     }
@@ -198,6 +229,7 @@ public FilteredForestConfiguration withBlackList(String... hostNames) {
    */
   public FilteredForestConfiguration withWhiteList(String... hostNames) {
     if ( blackList.size() > 0 ) throw new IllegalStateException("blackList already initialized");
+    if ( hostNames == null ) throw new IllegalArgumentException("hostNames must not be null");
     for ( String hostName : hostNames ) {
       if ( hostName != null ) whiteList.add(hostName.toLowerCase());
     }

src/main/java/com/marklogic/client/datamovement/Forest.java

@@ -60,6 +60,14 @@ public interface Forest {
    */
   String getAlternateHost();
 
+  /** The request host name associated with the forest in the MarkLogic server.
+   * The host name with which we have created the client - host name on the 
+   * request header.
+   *
+   * @return the host name of the request host in the MarkLogic server
+   */
+  String getRequestHost();
+
   /** Whether or not this forest is updateable.
    *
    * @return whether or not this forest is updateable
@@ -69,7 +77,8 @@ public interface Forest {
   /**
    * Returns the host your application should talk to for this forest.  If
    * getAlternateHost() is not null, return it.  Otherwise, if
-   * getOpenReplicaHost() is not null, return it.  Otherwise, return getHost().
+   * getOpenReplicaHost() is not null, return it. Otherwise, if
+   * getRequestHost() is not null, return it. Otherwise, return getHost().
    *
    * @return the preferred host for this forest
    */
@@ -78,8 +87,48 @@ public default String getPreferredHost() {
       return getAlternateHost();
     } else if ( getOpenReplicaHost() != null ) {
       return getOpenReplicaHost();
+    } else if ( getRequestHost() != null ) {
+      return getRequestHost();
     } else {
       return getHost();
     }
   }
+
+  /**
+   * Enum containing the list of host types a forest can have.
+   * 
+   * FOREST_HOST - host type indicating the host associated with 
+   * the forest in the MarkLogic server
+   * 
+   * REQUEST_HOST - host type indicating the request host with which 
+   * we have created the client - host on the request header.
+   *
+   * ALTERNATE_HOST - host type indicating the alternate host of the forest -
+   * the host of the e - node when this forest belongs to an d - node.
+   * 
+   * OPEN_REPLICA_HOST - host type indicating the replica host associated
+   * with the forest in the MarkLogic server
+   */
+  enum HostType { FOREST_HOST, REQUEST_HOST, ALTERNATE_HOST, OPEN_REPLICA_HOST };
+
+  /**
+   * Returns the preferred host type for this forest.  If
+   * getAlternateHost() is not null, return ALTERNATE_HOST. Otherwise, if
+   * getOpenReplicaHost() is not null, return OPEN_REPLICA_HOST. Otherwise, 
+   * if getRequestHost() is  not null, return REQUEST_HOST. 
+   * Otherwise, return FOREST_HOST.
+   * 
+   * @return the host type of the forest
+   */
+  public default HostType getPreferredHostType() {
+    if ( getAlternateHost() != null ) {
+      return HostType.ALTERNATE_HOST;
+    } else if ( getOpenReplicaHost() != null ) {
+      return HostType.OPEN_REPLICA_HOST;
+    } else if ( getRequestHost() != null ) {
+      return HostType.REQUEST_HOST;
+    } else {
+      return HostType.FOREST_HOST;
+    }
+  }
 }

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

@@ -18,6 +18,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.marklogic.client.datamovement.Forest.HostType;
+
 import java.net.SocketException;
 import java.net.UnknownHostException;
 import javax.net.ssl.SSLException;
@@ -189,7 +191,19 @@ private synchronized boolean processException(Batcher batcher, Throwable throwab
     boolean shouldWeRetry = isHostUnavailableException;
     if ( isHostUnavailableException == true ) {
       ForestConfiguration existingForestConfig = batcher.getForestConfig();
-      String[] preferredHosts = existingForestConfig.getPreferredHosts();
+      Set<String> preferredHostsList = new HashSet<String>(Arrays.asList(existingForestConfig.getPreferredHosts()));
+      if(existingForestConfig instanceof FilteredForestConfiguration) {
+        FilteredForestConfiguration existingFilteredForestConfiguration = (FilteredForestConfiguration) existingForestConfig;
+        for( Forest forest : existingFilteredForestConfiguration.listForests()) {
+          if(forest.getPreferredHostType() == HostType.REQUEST_HOST
+              && !forest.getHost().toLowerCase().equals(forest.getRequestHost().toLowerCase())) {
+            if(preferredHostsList.contains(forest.getHost())) {
+              preferredHostsList.remove(forest.getHost());
+            }
+          }
+        }
+      }
+      String[] preferredHosts = preferredHostsList.toArray(new String[preferredHostsList.size()]);
       if ( ! Arrays.asList(preferredHosts).contains(host) ) {
         // skip all the logic below because the host in question here is already
         // missing from the list of hosts for this batcher

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

@@ -169,12 +169,7 @@ public ForestConfiguration readForestConfig() {
 
   public DatabaseClient getForestClient(Forest forest) {
     if ( forest == null ) throw new IllegalArgumentException("forest must not be null");
-    String hostName = forest.getHost();
-    if ( forest.getOpenReplicaHost() != null ) {
-      hostName = forest.getOpenReplicaHost();
-    } else if ( forest.getAlternateHost() != null ) {
-      hostName = forest.getAlternateHost();
-    }
+    String hostName = forest.getPreferredHost();
     String key = hostName;
     DatabaseClient client = clientMap.get(key);
     if ( client != null ) return client;

src/main/java/com/marklogic/client/datamovement/impl/DataMovementServices.java

@@ -57,11 +57,18 @@ public ForestConfigurationImpl readForestConfig() {
       String host = forestNode.get("host").asText();
       String openReplicaHost = null;
       if ( forestNode.get("openReplicaHost") != null ) openReplicaHost = forestNode.get("openReplicaHost").asText();
+      String requestHost = null;
+      if ( forestNode.get("requestHost") != null ) requestHost = forestNode.get("requestHost").asText();
       String alternateHost = null;
       if ( forestNode.get("alternateHost") != null ) alternateHost = forestNode.get("alternateHost").asText();
+      // Since we added the forestinfo end point to populate both alternateHost and requestHost
+      // in case we have a requestHost so that we don't break the existing API code, we will make the 
+      // alternateHost as null if both alternateHost and requestHost is set.
+      if ( requestHost != null && alternateHost != null )
+        alternateHost = null;
       boolean isUpdateable = "all".equals(forestNode.get("updatesAllowed").asText());
       boolean isDeleteOnly = false; // TODO: get this for real after we start using a REST endpoint
-      forests.add(new ForestImpl(host, alternateHost, openReplicaHost, database, name, id, isUpdateable, isDeleteOnly));
+      forests.add(new ForestImpl(host, openReplicaHost, requestHost, alternateHost, database, name, id, isUpdateable, isDeleteOnly));
     }
 
     return new ForestConfigurationImpl(forests.toArray(new ForestImpl[forests.size()]));

src/main/java/com/marklogic/client/datamovement/impl/ForestImpl.java

@@ -21,18 +21,20 @@ public class ForestImpl implements Forest {
   private String host;
   private String openReplicaHost;
   private String alternateHost;
+  private String requestHost;
   private String databaseName;
   private String forestName;
   private String forestId;
   private boolean isUpdateable;
   private boolean isDeleteOnly;
 
-  public ForestImpl(String host, String openReplicaHost, String alternateHost, String databaseName,
+  public ForestImpl(String host, String openReplicaHost, String requestHost, String alternateHost, String databaseName,
     String forestName, String forestId, boolean isUpdateable, boolean isDeleteOnly)
   {
     this.host = host;
     this.openReplicaHost = openReplicaHost;
     this.alternateHost = alternateHost;
+    this.requestHost = requestHost;
     this.databaseName  = databaseName;
     this.forestName    = forestName;
     this.forestId      = forestId;
@@ -67,6 +69,15 @@ public void setAlternateHost(String alternateHost) {
     this.alternateHost = alternateHost;
   }
 
+  @Override
+  public String getRequestHost() {
+    return requestHost;
+  }
+
+  public void setRequestHost(String requestHost) {
+    this.requestHost = requestHost;
+  }
+
   @Override
   public String getDatabaseName() {
     return databaseName;