Merge pull request JanusGraph#1331 from jerryjch/merge-to-master

Merge PR JanusGraph#1282 to master

Author: jerryjch

janusgraph-hadoop-parent/janusgraph-hadoop-core/src/main/java/org/janusgraph/hadoop/MapReduceIndexManagement.java

@@ -54,6 +54,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
 import java.util.EnumSet;
 import java.util.Iterator;
 import java.util.Set;
@@ -177,7 +178,7 @@ public JanusGraphManagement.IndexJobFuture updateIndex(Index index, SchemaAction
         janusGraphMapReduceConfiguration.set(JanusGraphHadoopConfiguration.SCAN_JOB_CONFIG_ROOT,
                 GraphDatabaseConfiguration.class.getName() + "#JOB_NS");
         // Copy the StandardJanusGraph configuration under JanusGraphHadoopConfiguration.GRAPH_CONFIG_KEYS
-        org.apache.commons.configuration.Configuration localConfiguration = graph.getConfiguration().getLocalConfiguration();
+        org.apache.commons.configuration.Configuration localConfiguration = graph.getConfiguration().getConfigurationAtOpen();
         localConfiguration.clearProperty(Graph.GRAPH);
         copyInputKeys(hadoopConf, localConfiguration);
 
@@ -192,22 +193,18 @@ public JanusGraphManagement.IndexJobFuture updateIndex(Index index, SchemaAction
 
     private static void copyInputKeys(org.apache.hadoop.conf.Configuration hadoopConf, org.apache.commons.configuration.Configuration source) {
         // Copy IndexUpdateJob settings into the hadoop-backed cfg
-        Iterator<String> iterator = source.getKeys();
-        while (iterator.hasNext()) {
-            String key = iterator.next();
-            ConfigElement.PathIdentifier pid;
-            try {
-                pid = ConfigElement.parse(ROOT_NS, key);
-            } catch (RuntimeException e) {
-                log.debug("[inputkeys] Skipping {}", key, e);
-                continue;
-            }
-
-            if (!pid.element.isOption())
-                continue;
+        Iterator<String> keyIter = source.getKeys();
+        while (keyIter.hasNext()) {
+            String key = keyIter.next();
 
             String k = ConfigElement.getPath(JanusGraphHadoopConfiguration.GRAPH_CONFIG_KEYS, true) + "." + key;
-            String v = source.getProperty(key).toString();
+            Object vObject = source.getProperty(key);
+            String v;
+            if (vObject instanceof Collection) {
+                v = Joiner.on(",").join((Collection<String>) vObject);
+            } else {
+                v = vObject.toString();
+            }
 
             hadoopConf.set(k, v);
             log.debug("[inputkeys] Set {}={}", k, v);

janusgraph-solr/src/main/java/org/janusgraph/diskstorage/solr/SolrIndex.java

@@ -26,6 +26,7 @@
 import java.time.Instant;
 import java.util.AbstractMap.SimpleEntry;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
@@ -168,10 +169,13 @@ public static Mode parse(String mode) {
             ConfigOption.Type.GLOBAL_OFFLINE, "ttl");
 
     /** SolrCloud Configuration */
-
-    public static final ConfigOption<String> ZOOKEEPER_URL = new ConfigOption<>(SOLR_NS,"zookeeper-url",
+    /*
+     * TODO Rename ZOOKEEPER_URL and "zookeeper-url" to ZOOKEEPER_URLS and
+     * "zookeeper-urls" in future major releases.
+     */
+    public static final ConfigOption<String[]> ZOOKEEPER_URL = new ConfigOption<String[]>(SOLR_NS,"zookeeper-url",
             "URL of the Zookeeper instance coordinating the SolrCloud cluster",
-            ConfigOption.Type.MASKABLE, "localhost:2181");
+            ConfigOption.Type.MASKABLE, new String[] { "localhost:2181" });
 
     public static final ConfigOption<Integer> NUM_SHARDS = new ConfigOption<>(SOLR_NS,"num-shards",
             "Number of shards for a collection. This applies when creating a new collection which is only supported under the SolrCloud operation mode.",
@@ -268,18 +272,36 @@ public SolrIndex(final Configuration config) throws BackendException {
         final ModifiableSolrParams clientParams = new ModifiableSolrParams();
         switch (mode) {
             case CLOUD:
-                final String zookeeperUrl = config.get(SolrIndex.ZOOKEEPER_URL);
-                final CloudSolrClient cloudServer = new CloudSolrClient.Builder()
+                final String[] zookeeperUrl = config.get(SolrIndex.ZOOKEEPER_URL);
+                // Process possible zookeeper chroot. e.g. localhost:2181/solr
+                // chroot has to be the same assuming one Zookeeper ensemble.
+                // Parse from the last string. If found, take it as the chroot.
+                String chroot = null;
+                for (int i = zookeeperUrl.length - 1; i >= 0; i--) {
+                    int chrootIndex = zookeeperUrl[i].indexOf("/");
+                    if (chrootIndex != -1) {
+                        String hostAndPort = zookeeperUrl[i].substring(0, chrootIndex);
+                        if (chroot == null) {
+                            chroot = zookeeperUrl[i].substring(chrootIndex);
+                        }
+                        zookeeperUrl[i] = hostAndPort;
+                    }
+                }
+                final CloudSolrClient.Builder builder = new CloudSolrClient.Builder()
                     .withLBHttpSolrClientBuilder(
                         new LBHttpSolrClient.Builder()
                             .withHttpSolrClientBuilder(new HttpSolrClient.Builder().withInvariantParams(clientParams))
                             .withBaseSolrUrls(config.get(HTTP_URLS))
-                     )
-                    .withZkHost(zookeeperUrl)
-                    .sendUpdatesOnlyToShardLeaders()
-                    .build();
+                         )
+                    .withZkHost(Arrays.asList(zookeeperUrl))
+                    .sendUpdatesOnlyToShardLeaders();
+                if (chroot != null) {
+                    builder.withZkChroot(chroot);
+                }
+                final CloudSolrClient cloudServer = builder.build();
                 cloudServer.connect();
                 solrClient = cloudServer;
+
                 break;
             case HTTP:
                 clientParams.add(HttpClientUtil.PROP_ALLOW_COMPRESSION, config.get(HTTP_ALLOW_COMPRESSION).toString());
@@ -292,7 +314,6 @@ public SolrIndex(final Configuration config) throws BackendException {
                     .withBaseSolrUrls(config.get(HTTP_URLS))
                     .build();
 
-
                 break;
             default:
                 throw new IllegalArgumentException("Unsupported Solr operation mode: " + mode);
@@ -1208,4 +1229,4 @@ private static void waitForRecoveriesToFinish(CloudSolrClient server, String col
         }
     }
 
-}
+}

janusgraph-solr/src/test/java/org/janusgraph/diskstorage/solr/SolrRunner.java

@@ -16,6 +16,8 @@
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
 import org.apache.commons.io.FileUtils;
 import org.apache.solr.client.solrj.embedded.JettyConfig;
 import org.apache.solr.cloud.ConfigurableMiniSolrCloudCluster;
@@ -34,7 +36,7 @@
 
 public class SolrRunner {
 
-    public static final String ZOOKEEPER_URLS = System.getProperty("index.search.solr.zookeeper-url");
+    public static final String ZOOKEEPER_URLS_SYSTEM_PROPERTY = System.getProperty("index.search.solr.zookeeper-url");
 
     protected static final int NUM_SERVERS = 1;
     protected static final String[] COLLECTIONS = readCollections();
@@ -58,7 +60,7 @@ public static void start() throws Exception {
     public static void start(boolean isKerberosEnabled) throws Exception {
         kerberosEnabled = isKerberosEnabled;
 
-        if (ZOOKEEPER_URLS != null) {
+        if (ZOOKEEPER_URLS_SYSTEM_PROPERTY != null) {
             return;
         }
         String userDir = System.getProperty("user.dir");
@@ -98,18 +100,18 @@ public static void start(boolean isKerberosEnabled) throws Exception {
         }
     }
 
-	public static String getZookeeperUrls() {
-        final String zookeeperUrls;
-        if (ZOOKEEPER_URLS == null) {
-            zookeeperUrls = miniSolrCloudCluster.getZkServer().getZkAddress();
+    public static String[] getZookeeperUrls() {
+        final String[] zookeeperUrls;
+        if (Strings.isNullOrEmpty(ZOOKEEPER_URLS_SYSTEM_PROPERTY)) {
+            zookeeperUrls = new String[] { miniSolrCloudCluster.getZkServer().getZkAddress() };
         } else {
-            zookeeperUrls = ZOOKEEPER_URLS;
+            zookeeperUrls = ZOOKEEPER_URLS_SYSTEM_PROPERTY.split(",");
         }
         return zookeeperUrls;
     }
 
     public static void stop() throws Exception {
-        if (ZOOKEEPER_URLS != null) {
+        if (ZOOKEEPER_URLS_SYSTEM_PROPERTY != null) {
             return;
         }
         System.clearProperty("solr.solrxml.location");