Repository: atlas Updated Branches: refs/heads/branch-0.8 01f3f14ea -> 2f3942831
ATLAS-2786: updated HBase and Solr clients to use all zookeeper entires Project: http://git-wip-us.apache.org/repos/asf/atlas/repo Commit: http://git-wip-us.apache.org/repos/asf/atlas/commit/2f394283 Tree: http://git-wip-us.apache.org/repos/asf/atlas/tree/2f394283 Diff: http://git-wip-us.apache.org/repos/asf/atlas/diff/2f394283 Branch: refs/heads/branch-0.8 Commit: 2f39428313136b765dbab42a1ee2e6521695d690 Parents: 01f3f14 Author: Madhan Neethiraj <mad...@apache.org> Authored: Thu Sep 27 19:51:35 2018 -0700 Committer: Madhan Neethiraj <mad...@apache.org> Committed: Thu Sep 27 21:06:09 2018 -0700 ---------------------------------------------------------------------- .../titan/diskstorage/solr/Solr5Index.java | 50 +++++++++++++++++--- .../audit/HBaseBasedAuditRepository.java | 20 ++++---- 2 files changed, 55 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/atlas/blob/2f394283/graphdb/titan0/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java ---------------------------------------------------------------------- diff --git a/graphdb/titan0/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java b/graphdb/titan0/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java index 90d24e4..50b3123 100644 --- a/graphdb/titan0/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java +++ b/graphdb/titan0/src/main/java/com/thinkaurelius/titan/diskstorage/solr/Solr5Index.java @@ -103,7 +103,8 @@ public class Solr5Index implements IndexProvider { private static final Logger logger = LoggerFactory.getLogger(Solr5Index.class); - private static final String DEFAULT_ID_FIELD = "id"; + private static final String DEFAULT_ID_FIELD = "id"; + private static final char CHROOT_START_CHAR = '/'; private enum Mode { HTTP, CLOUD; @@ -138,9 +139,9 @@ public class Solr5Index implements IndexProvider { /** SolrCloud Configuration */ - public static final ConfigOption<String> ZOOKEEPER_URL = new ConfigOption<>(SOLR_NS, "zookeeper-url", + public static final ConfigOption<String[]> ZOOKEEPER_URL = new ConfigOption<>(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> ZOOKEEPER_CONNECT_TIMEOUT = new ConfigOption<>(SOLR_NS,"zookeeper-connect-timeout", "SolrCloud Zookeeper connect timeout", @@ -216,8 +217,9 @@ public class Solr5Index implements IndexProvider { if (mode==Mode.CLOUD) { HttpClientUtil.setConfigurer(new Krb5HttpClientConfigurer()); - String zookeeperUrl = config.get(Solr5Index.ZOOKEEPER_URL); - CloudSolrClient cloudServer = new CloudSolrClient(zookeeperUrl, true); + + String[] zkHosts = config.get(Solr5Index.ZOOKEEPER_URL); + CloudSolrClient cloudServer = getCloudSolrClient(zkHosts); logger.info("Zookeeper connect timeout : " + config.get(ZOOKEEPER_CONNECT_TIMEOUT)); cloudServer.setZkConnectTimeout(config.get(ZOOKEEPER_CONNECT_TIMEOUT)); @@ -237,8 +239,6 @@ public class Solr5Index implements IndexProvider { }}); solrClient = new LBHttpSolrClient(clientParams, config.get(HTTP_URLS)); - - } else { throw new IllegalArgumentException("Unsupported Solr operation mode: " + mode); } @@ -966,6 +966,42 @@ public class Solr5Index implements IndexProvider { } } + private CloudSolrClient getCloudSolrClient(String[] zkHosts) { + logger.info("==> Solr5Index.getCloudSolrClient({})", zkHosts); + + CloudSolrClient ret = null; + + if (zkHosts != null) { + List<String> zkServers = new ArrayList<>(zkHosts.length); + String chroot = null; + + for (String zkHost : zkHosts) { + if (zkHost != null) { + int idxSlash = zkHost.indexOf(CHROOT_START_CHAR); + + if (idxSlash != -1) { + chroot = zkHost.substring(idxSlash); // chroot of the last url will be used + zkHost = zkHost.substring(0, idxSlash); + } + + if (StringUtils.isNotEmpty(zkHost)) { + logger.info("Solr5Index: adding zkHost={}. chroot={}", zkHost, chroot); + + zkServers.add(zkHost); + } + } + } + + logger.info("Solr5Index: number of zkUrls={}. chroot={}", zkServers.size(), chroot); + + ret = new CloudSolrClient(zkServers, chroot); + } + + logger.info("<== Solr5Index.getCloudSolrClient({}): ret={}", zkHosts, ret); + + return ret; + } + private static class GeoToWktConverter { /** * {@link com.thinkaurelius.titan.core.attribute.Geoshape} stores Points in the String format: point[X.0,Y.0]. http://git-wip-us.apache.org/repos/asf/atlas/blob/2f394283/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java ---------------------------------------------------------------------- diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java index 774934c..ea88751 100644 --- a/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java +++ b/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java @@ -27,6 +27,7 @@ import org.apache.atlas.ha.HAConfiguration; import org.apache.atlas.listener.ActiveStateChangeHandler; import org.apache.atlas.service.Service; import org.apache.commons.configuration.Configuration; +import org.apache.commons.configuration.ConfigurationConverter; import org.apache.commons.lang.StringUtils; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HColumnDescriptor; @@ -55,9 +56,9 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; -import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Properties; /** * HBase based repository for entity audit events @@ -325,13 +326,16 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository * @param atlasConf */ public static org.apache.hadoop.conf.Configuration getHBaseConfiguration(Configuration atlasConf) throws AtlasException { - Configuration subsetAtlasConf = - ApplicationProperties.getSubsetConfiguration(atlasConf, CONFIG_PREFIX); - org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create(); - Iterator<String> keys = subsetAtlasConf.getKeys(); - while (keys.hasNext()) { - String key = keys.next(); - hbaseConf.set(key, subsetAtlasConf.getString(key)); + Configuration subsetAtlasConf = ApplicationProperties.getSubsetConfiguration(atlasConf, CONFIG_PREFIX); + Properties properties = ConfigurationConverter.getProperties(subsetAtlasConf); + org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create(); + + for (String key : properties.stringPropertyNames()) { + String value = properties.getProperty(key); + + LOG.info("adding HBase configuration: {}={}", key, value); + + hbaseConf.set(key, value); } return hbaseConf; }
