You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2017/05/19 10:05:02 UTC
[35/37] kylin git commit: KYLIN-2624 introduce StorageURL and set
timeout for metadata access
KYLIN-2624 introduce StorageURL and set timeout for metadata access
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/2dfc5bb3
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/2dfc5bb3
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/2dfc5bb3
Branch: refs/heads/master
Commit: 2dfc5bb36a00245b95deeebb5bb3bca9c353c934
Parents: 7c152c5
Author: Li Yang <li...@apache.org>
Authored: Thu May 18 14:16:17 2017 +0800
Committer: nichunen <zj...@sjtu.org>
Committed: Fri May 19 13:52:53 2017 +0800
----------------------------------------------------------------------
.../apache/kylin/common/KylinConfigBase.java | 30 ++-
.../org/apache/kylin/common/StorageURL.java | 183 +++++++++++++++++++
.../common/persistence/FileResourceStore.java | 2 +-
.../apache/kylin/common/KylinConfigTest.java | 9 +-
.../org/apache/kylin/common/StorageURLTest.java | 103 +++++++++++
.../persistence/LocalFileResourceStoreTest.java | 3 +-
.../common/persistence/ResourceStoreTest.java | 18 +-
.../common/util/LocalFileMetadataTestCase.java | 7 +
.../apache/kylin/dict/DictionaryManager.java | 18 +-
.../dict/global/AppendTrieDictionaryTest.java | 5 +-
.../impl/threadpool/DistributedScheduler.java | 11 +-
.../apache/kylin/storage/StorageContext.java | 7 +-
.../kylin/engine/mr/steps/CubeReducerTest.java | 2 +-
.../engine/mr/steps/NDCuboidMapperTest.java | 2 +-
.../kylin/job/BaseTestDistributedScheduler.java | 5 +-
.../kylin/storage/hbase/ITStorageTest.java | 3 +-
.../apache/kylin/query/schema/OLAPSchema.java | 5 +-
.../rest/security/MockAclHBaseStorage.java | 11 +-
.../rest/security/RealAclHBaseStorage.java | 15 +-
.../apache/kylin/rest/service/QueryService.java | 11 +-
.../kylin/storage/hbase/HBaseConnection.java | 51 +++---
.../kylin/storage/hbase/HBaseResourceStore.java | 46 +++--
.../storage/hbase/util/CubeMigrationCLI.java | 20 +-
.../hbase/util/GridTableHBaseBenchmark.java | 3 +-
.../kylin/storage/hdfs/HDFSResourceStore.java | 50 ++---
.../hbase/steps/CubeHFileMapper2Test.java | 4 +-
.../storage/hbase/steps/MockupMapContext.java | 2 +-
.../org/apache/kylin/tool/CubeMigrationCLI.java | 20 +-
28 files changed, 461 insertions(+), 185 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index b620924..caef95e 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -169,7 +169,7 @@ abstract public class KylinConfigBase implements Serializable {
}
public String toString() {
- return getMetadataUrl();
+ return getMetadataUrl().toString();
}
// ============================================================================
@@ -205,8 +205,8 @@ abstract public class KylinConfigBase implements Serializable {
// METADATA
// ============================================================================
- public String getMetadataUrl() {
- return getOptional("kylin.metadata.url");
+ public StorageURL getMetadataUrl() {
+ return StorageURL.valueOf(getOptional("kylin.metadata.url", ""));
}
// for test only
@@ -215,14 +215,7 @@ abstract public class KylinConfigBase implements Serializable {
}
public String getMetadataUrlPrefix() {
- String metadataUrl = getMetadataUrl();
- String defaultPrefix = "kylin_metadata";
-
- int cut = metadataUrl.lastIndexOf('@');
- if (cut > 0) {
- return metadataUrl.substring(0, cut);
- }
- return defaultPrefix;
+ return getMetadataUrl().getIdentifier();
}
public String[] getRealizationProviders() {
@@ -558,8 +551,14 @@ abstract public class KylinConfigBase implements Serializable {
return Integer.parseInt(getOptional("kylin.storage.default", "2"));
}
- public String getStorageUrl() {
- return getOptional("kylin.storage.url");
+ public StorageURL getStorageUrl() {
+ String url = getOptional("kylin.storage.url", "default@hbase");
+
+ // for backward compatibility
+ if ("hbase".equals(url))
+ url = "default@hbase";
+
+ return StorageURL.valueOf(url);
}
public String getHBaseClusterFs() {
@@ -987,10 +986,7 @@ abstract public class KylinConfigBase implements Serializable {
}
public String getResourceStoreImpl() {
- String metadataUrl = KylinConfig.getInstanceFromEnv().getMetadataUrl();
- int cut = metadataUrl.indexOf('@');
- String key = cut < 0 ? "" : metadataUrl.substring(cut + 1);
- return getResourceStoreImpls().get(key);
+ return getResourceStoreImpls().get(getMetadataUrl().getScheme());
}
public String getJobTrackingURLPattern() {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/main/java/org/apache/kylin/common/StorageURL.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/StorageURL.java b/core-common/src/main/java/org/apache/kylin/common/StorageURL.java
new file mode 100644
index 0000000..cebbc27
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/StorageURL.java
@@ -0,0 +1,183 @@
+/*
+ * 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.kylin.common;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableMap;
+
+/**
+ * The object form of metadata/storage URL: IDENTIFIER@SCHEME[,PARAM=VALUE,PARAM=VALUE...]
+ *
+ * It is not standard URL, but a string of specific format that shares some similar parts with URL.
+ *
+ * Immutable by design.
+ */
+public class StorageURL {
+
+ private static final LoadingCache<String, StorageURL> cache = CacheBuilder.newBuilder()//
+ .maximumSize(100)//
+ .build(new CacheLoader<String, StorageURL>() {
+ @Override
+ public StorageURL load(String metadataUrl) throws Exception {
+ return new StorageURL(metadataUrl);
+ }
+ });
+
+ public static StorageURL valueOf(String metadataUrl) {
+ try {
+ return cache.get(metadataUrl);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // ============================================================================
+
+ final String identifier;
+ final String scheme;
+ final Map<String, String> params;
+
+ // package private for test
+ StorageURL(String metadataUrl) {
+ boolean first = true;
+ String n = null;
+ String s = null;
+ Map<String, String> m = new LinkedHashMap<>();
+
+ for (String split : metadataUrl.split(",")) {
+ if (first) {
+ // identifier @ scheme
+ int cut = split.lastIndexOf('@');
+ if (cut < 0) {
+ n = split.trim();
+ s = "";
+ } else {
+ n = split.substring(0, cut).trim();
+ s = split.substring(cut + 1).trim();
+ }
+ first = false;
+ } else {
+ // param = value
+ int cut = split.indexOf('=');
+ String k, v;
+ if (cut < 0) {
+ k = split.trim();
+ v = "";
+ } else {
+ k = split.substring(0, cut).trim();
+ v = split.substring(cut + 1).trim();
+ }
+ m.put(k, v);
+ }
+ }
+
+ this.identifier = n.isEmpty() ? "kylin_metadata" : n;
+ this.scheme = s;
+ this.params = ImmutableMap.copyOf(m);
+ }
+
+ private StorageURL(String identifier, String scheme, Map<String, String> params) {
+ this.identifier = identifier;
+ this.scheme = scheme;
+ this.params = ImmutableMap.copyOf(params);
+ }
+
+ public String getIdentifier() {
+ return identifier;
+ }
+
+ public String getScheme() {
+ return scheme;
+ }
+
+ public boolean containsParameter(String k) {
+ return params.containsKey(k);
+ }
+
+ public String getParameter(String k) {
+ return params.get(k);
+ }
+
+ public Map<String, String> getAllParameters() {
+ return params;
+ }
+
+ public StorageURL copy(Map<String, String> params) {
+ return new StorageURL(identifier, scheme, params);
+ }
+
+ @Override
+ public String toString() {
+ String str = identifier;
+ if (!scheme.isEmpty())
+ str += "@" + scheme;
+
+ for (Entry<String, String> kv : params.entrySet()) {
+ str += "," + kv.getKey();
+ if (!kv.getValue().isEmpty())
+ str += "=" + kv.getValue();
+ }
+ return str;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((identifier == null) ? 0 : identifier.hashCode());
+ result = prime * result + ((params == null) ? 0 : params.hashCode());
+ result = prime * result + ((scheme == null) ? 0 : scheme.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ StorageURL other = (StorageURL) obj;
+ if (identifier == null) {
+ if (other.identifier != null)
+ return false;
+ } else if (!identifier.equals(other.identifier))
+ return false;
+ if (params == null) {
+ if (other.params != null)
+ return false;
+ } else if (!params.equals(other.params))
+ return false;
+ if (scheme == null) {
+ if (other.scheme != null)
+ return false;
+ } else if (!scheme.equals(other.scheme))
+ return false;
+ return true;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java b/core-common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
index d84e587..82cf451 100644
--- a/core-common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
+++ b/core-common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
@@ -44,7 +44,7 @@ public class FileResourceStore extends ResourceStore {
public FileResourceStore(KylinConfig kylinConfig) {
super(kylinConfig);
- root = new File(kylinConfig.getMetadataUrl()).getAbsoluteFile();
+ root = new File(kylinConfig.getMetadataUrl().getIdentifier()).getAbsoluteFile();
if (root.exists() == false)
throw new IllegalArgumentException("File not exist by '" + kylinConfig.getMetadataUrl() + "': " + root.getAbsolutePath());
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/test/java/org/apache/kylin/common/KylinConfigTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/KylinConfigTest.java b/core-common/src/test/java/org/apache/kylin/common/KylinConfigTest.java
index 3976c6c..6027af3 100644
--- a/core-common/src/test/java/org/apache/kylin/common/KylinConfigTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/KylinConfigTest.java
@@ -86,7 +86,6 @@ public class KylinConfigTest extends HotLoadKylinPropertiesTestCase {
@Test
public void testGetMetadataUrlPrefix() {
KylinConfig config = KylinConfig.getInstanceFromEnv();
- final String default_metadata_prefix = "kylin_metadata";
config.setMetadataUrl("testMetaPrefix@hbase");
assertEquals("testMetaPrefix", config.getMetadataUrlPrefix());
@@ -95,7 +94,7 @@ public class KylinConfigTest extends HotLoadKylinPropertiesTestCase {
assertEquals("testMetaPrefix", config.getMetadataUrlPrefix());
config.setMetadataUrl("/kylin/temp");
- assertEquals(default_metadata_prefix, config.getMetadataUrlPrefix());
+ assertEquals("/kylin/temp", config.getMetadataUrlPrefix());
}
@Test
@@ -107,21 +106,21 @@ public class KylinConfigTest extends HotLoadKylinPropertiesTestCase {
KylinConfig sysConfig = KylinConfig.getInstanceFromEnv();
sysConfig.setMetadataUrl(metadata1);
- assertEquals(metadata1, KylinConfig.getInstanceFromEnv().getMetadataUrl());
+ assertEquals(metadata1, KylinConfig.getInstanceFromEnv().getMetadataUrl().toString());
// test thread-local override
KylinConfig threadConfig = KylinConfig.createKylinConfig(new Properties());
threadConfig.setMetadataUrl(metadata2);
KylinConfig.setKylinConfigThreadLocal(threadConfig);
- assertEquals(metadata2, KylinConfig.getInstanceFromEnv().getMetadataUrl());
+ assertEquals(metadata2, KylinConfig.getInstanceFromEnv().getMetadataUrl().toString());
// other threads still use system KylinConfig
new Thread(new Runnable() {
@Override
public void run() {
System.out.println("Started new thread.");
- assertEquals(metadata1, KylinConfig.getInstanceFromEnv().getMetadataUrl());
+ assertEquals(metadata1, KylinConfig.getInstanceFromEnv().getMetadataUrl().toString());
}
}).start();
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/test/java/org/apache/kylin/common/StorageURLTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/StorageURLTest.java b/core-common/src/test/java/org/apache/kylin/common/StorageURLTest.java
new file mode 100644
index 0000000..eaa7548
--- /dev/null
+++ b/core-common/src/test/java/org/apache/kylin/common/StorageURLTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.kylin.common;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class StorageURLTest {
+
+ @Test
+ public void testBasic() {
+ {
+ StorageURL id = new StorageURL("hello@hbase");
+ assertEquals("hello", id.getIdentifier());
+ assertEquals("hbase", id.getScheme());
+ assertEquals(0, id.getAllParameters().size());
+ assertEquals("hello@hbase", id.toString());
+ }
+ {
+ StorageURL id = new StorageURL("hello@hbase,a=b,c=d");
+ assertEquals("hello", id.getIdentifier());
+ assertEquals("hbase", id.getScheme());
+ assertEquals(2, id.getAllParameters().size());
+ assertEquals("b", id.getParameter("a"));
+ assertEquals("d", id.getParameter("c"));
+ assertEquals("hello@hbase,a=b,c=d", id.toString());
+ }
+ {
+ StorageURL o = new StorageURL("hello@hbase,c=d");
+ StorageURL o2 = new StorageURL("hello@hbase,a=b");
+ StorageURL id = o.copy(o2.getAllParameters());
+ assertEquals("hello", id.getIdentifier());
+ assertEquals("hbase", id.getScheme());
+ assertEquals(1, id.getAllParameters().size());
+ assertEquals("b", id.getParameter("a"));
+ assertEquals("hello@hbase,a=b", id.toString());
+ assertEquals("hello@hbase,c=d", o.toString());
+ assertEquals("hello@hbase,a=b", o2.toString());
+ }
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testNullInput() {
+ new StorageURL(null);
+ }
+
+ @Test
+ public void testEdgeCases() {
+ {
+ StorageURL id = new StorageURL("");
+ assertEquals("kylin_metadata", id.getIdentifier());
+ assertEquals("", id.getScheme());
+ assertEquals(0, id.getAllParameters().size());
+ assertEquals("kylin_metadata", id.toString());
+ }
+ {
+ StorageURL id = new StorageURL("hello@");
+ assertEquals("hello", id.getIdentifier());
+ assertEquals("", id.getScheme());
+ assertEquals(0, id.getAllParameters().size());
+ assertEquals("hello", id.toString());
+ }
+ {
+ StorageURL id = new StorageURL("hello@hbase,a");
+ assertEquals("hello", id.getIdentifier());
+ assertEquals("hbase", id.getScheme());
+ assertEquals(1, id.getAllParameters().size());
+ assertEquals("", id.getParameter("a"));
+ assertEquals("hello@hbase,a", id.toString());
+ }
+ }
+
+ @Test
+ public void testValueOfCache() {
+ StorageURL id1 = StorageURL.valueOf("hello@hbase");
+ StorageURL id2 = StorageURL.valueOf("hello@hbase");
+ StorageURL id3 = StorageURL.valueOf("hello @ hbase");
+ StorageURL id4 = StorageURL.valueOf("hello@hbase,a=b");
+ assertTrue(id1 == id2);
+ assertTrue(id1 != id3);
+ assertTrue(id1.equals(id3));
+ assertTrue(id2 != id4);
+ assertTrue(!id2.equals(id4));
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/test/java/org/apache/kylin/common/persistence/LocalFileResourceStoreTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/persistence/LocalFileResourceStoreTest.java b/core-common/src/test/java/org/apache/kylin/common/persistence/LocalFileResourceStoreTest.java
index aca4a0a..63eb04b 100644
--- a/core-common/src/test/java/org/apache/kylin/common/persistence/LocalFileResourceStoreTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/persistence/LocalFileResourceStoreTest.java
@@ -43,7 +43,8 @@ public class LocalFileResourceStoreTest extends LocalFileMetadataTestCase {
@Test
public void testFileStore() throws Exception {
- ResourceStoreTest.testAStore("", KylinConfig.getInstanceFromEnv());
+ KylinConfig config = KylinConfig.getInstanceFromEnv();
+ ResourceStoreTest.testAStore(config.getMetadataUrl().toString(), config);
}
@Test
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/test/java/org/apache/kylin/common/persistence/ResourceStoreTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/persistence/ResourceStoreTest.java b/core-common/src/test/java/org/apache/kylin/common/persistence/ResourceStoreTest.java
index 1f563b0..8aff1a3 100644
--- a/core-common/src/test/java/org/apache/kylin/common/persistence/ResourceStoreTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/persistence/ResourceStoreTest.java
@@ -18,19 +18,19 @@
package org.apache.kylin.common.persistence;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.ClassUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.List;
import java.util.NavigableSet;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.ClassUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Be called by LocalFileResourceStoreTest, ITHBaseResourceStoreTest and ITHDFSResourceStoreTest.
@@ -200,7 +200,7 @@ public class ResourceStoreTest {
}
public static String replaceMetadataUrl(KylinConfig kylinConfig, String newUrl) {
- String oldUrl = kylinConfig.getMetadataUrl();
+ String oldUrl = kylinConfig.getMetadataUrl().toString();
kylinConfig.setProperty("kylin.metadata.url", newUrl);
return oldUrl;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-common/src/test/java/org/apache/kylin/common/util/LocalFileMetadataTestCase.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/LocalFileMetadataTestCase.java b/core-common/src/test/java/org/apache/kylin/common/util/LocalFileMetadataTestCase.java
index 4a24941..e5e4f8b 100644
--- a/core-common/src/test/java/org/apache/kylin/common/util/LocalFileMetadataTestCase.java
+++ b/core-common/src/test/java/org/apache/kylin/common/util/LocalFileMetadataTestCase.java
@@ -76,6 +76,13 @@ public class LocalFileMetadataTestCase extends AbstractKylinTestCase {
public void cleanupTestMetadata() {
cleanAfterClass();
}
+
+ protected String getLocalWorkingDirectory() {
+ String dir = KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory();
+ if (dir.startsWith("file://"))
+ dir = dir.substring("file://".length());
+ return dir;
+ }
protected ResourceStore getStore() {
return ResourceStore.getStore(KylinConfig.getInstanceFromEnv());
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
index 427bd14..4c4e334 100644
--- a/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
+++ b/core-dictionary/src/main/java/org/apache/kylin/dict/DictionaryManager.java
@@ -90,13 +90,17 @@ public class DictionaryManager {
private DictionaryManager(KylinConfig config) {
this.config = config;
- this.dictCache = CacheBuilder.newBuilder().removalListener(new RemovalListener<String, DictionaryInfo>() {
- @Override
- public void onRemoval(RemovalNotification<String, DictionaryInfo> notification) {
- DictionaryManager.logger.info("Dict with resource path " + notification.getKey() + " is removed due to " + notification.getCause());
- }
- }).maximumSize(config.getCachedDictMaxEntrySize())//
- .expireAfterWrite(1, TimeUnit.DAYS).build(new CacheLoader<String, DictionaryInfo>() {
+ this.dictCache = CacheBuilder.newBuilder()//
+ .softValues()//
+ .removalListener(new RemovalListener<String, DictionaryInfo>() {
+ @Override
+ public void onRemoval(RemovalNotification<String, DictionaryInfo> notification) {
+ DictionaryManager.logger.info("Dict with resource path " + notification.getKey() + " is removed due to " + notification.getCause());
+ }
+ })//
+ .maximumSize(config.getCachedDictMaxEntrySize())//
+ .expireAfterWrite(1, TimeUnit.DAYS)//
+ .build(new CacheLoader<String, DictionaryInfo>() {
@Override
public DictionaryInfo load(String key) throws Exception {
DictionaryInfo dictInfo = DictionaryManager.this.load(key, true);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-dictionary/src/test/java/org/apache/kylin/dict/global/AppendTrieDictionaryTest.java
----------------------------------------------------------------------
diff --git a/core-dictionary/src/test/java/org/apache/kylin/dict/global/AppendTrieDictionaryTest.java b/core-dictionary/src/test/java/org/apache/kylin/dict/global/AppendTrieDictionaryTest.java
index f6ba75a..47011fe 100644
--- a/core-dictionary/src/test/java/org/apache/kylin/dict/global/AppendTrieDictionaryTest.java
+++ b/core-dictionary/src/test/java/org/apache/kylin/dict/global/AppendTrieDictionaryTest.java
@@ -46,8 +46,8 @@ import java.util.UUID;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
import org.apache.kylin.dict.AppendTrieDictionary;
import org.apache.kylin.dict.BytesConverter;
@@ -60,13 +60,14 @@ import org.junit.Test;
public class AppendTrieDictionaryTest extends LocalFileMetadataTestCase {
private static final String RESOURCE_DIR = "/dict/append_dict_test/" + UUID.randomUUID();
private static String BASE_DIR;
- private static String LOCAL_BASE_DIR = "/tmp/kylin/kylin_metadata/resources/GlobalDict" + RESOURCE_DIR + "/";
+ private static String LOCAL_BASE_DIR;
@Before
public void beforeTest() {
staticCreateTestMetadata();
KylinConfig.getInstanceFromEnv().setProperty("kylin.dictionary.append-entry-size", "50000");
BASE_DIR = KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "/resources/GlobalDict" + RESOURCE_DIR + "/";
+ LOCAL_BASE_DIR = getLocalWorkingDirectory() + "/resources/GlobalDict" + RESOURCE_DIR + "/";
}
@After
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DistributedScheduler.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DistributedScheduler.java b/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DistributedScheduler.java
index fa1b8e0..fd7b43a 100644
--- a/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DistributedScheduler.java
+++ b/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DistributedScheduler.java
@@ -320,13 +320,20 @@ public class DistributedScheduler implements Scheduler<AbstractExecutable>, Conn
}
public String getLockPath(String pathName) {
- return ZOOKEEPER_LOCK_PATH + "/" + KylinConfig.getInstanceFromEnv().getMetadataUrlPrefix() + "/" + pathName;
+ return dropDoubleSlash(ZOOKEEPER_LOCK_PATH + "/" + KylinConfig.getInstanceFromEnv().getMetadataUrlPrefix() + "/" + pathName);
}
private String getWatchPath() {
- return ZOOKEEPER_LOCK_PATH + "/" + KylinConfig.getInstanceFromEnv().getMetadataUrlPrefix();
+ return dropDoubleSlash(ZOOKEEPER_LOCK_PATH + "/" + KylinConfig.getInstanceFromEnv().getMetadataUrlPrefix());
}
+ public static String dropDoubleSlash(String path) {
+ for (int n = Integer.MAX_VALUE; n > path.length();) {
+ n = path.length();
+ path = path.replace("//", "/");
+ }
+ return path;
+ }
@Override
public void shutdown() throws SchedulerException {
logger.info("Will shut down Job Engine ....");
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
----------------------------------------------------------------------
diff --git a/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java b/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
index 4522261..78cf97c 100644
--- a/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
+++ b/core-storage/src/main/java/org/apache/kylin/storage/StorageContext.java
@@ -20,6 +20,7 @@ package org.apache.kylin.storage;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.metadata.realization.IRealization;
@@ -35,7 +36,7 @@ import com.google.common.collect.Range;
public class StorageContext {
private static final Logger logger = LoggerFactory.getLogger(StorageContext.class);
- private String connUrl;
+ private StorageURL connUrl;
private int limit = Integer.MAX_VALUE;
private boolean overlookOuterLimit = false;
private int offset = 0;
@@ -56,11 +57,11 @@ public class StorageContext {
private Range<Long> reusedPeriod;
- public String getConnUrl() {
+ public StorageURL getConnUrl() {
return connUrl;
}
- public void setConnUrl(String connUrl) {
+ public void setConnUrl(StorageURL connUrl) {
this.connUrl = connUrl;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
index 29ca9b8..7616df2 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
@@ -60,7 +60,7 @@ public class CubeReducerTest extends LocalFileMetadataTestCase {
// hack for distributed cache
FileUtils.deleteDirectory(new File("../job/meta"));
- FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl()), new File("../job/meta"));
+ FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl().toString()), new File("../job/meta"));
CuboidReducer reducer = new CuboidReducer();
reduceDriver = ReduceDriver.newReduceDriver(reducer);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
index 815d9b3..c0ce1a4 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/NDCuboidMapperTest.java
@@ -54,7 +54,7 @@ public class NDCuboidMapperTest extends LocalFileMetadataTestCase {
// hack for distributed cache
FileUtils.deleteDirectory(new File("./meta"));
- FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl()), new File("./meta"));
+ FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl().toString()), new File("./meta"));
NDCuboidMapper mapper = new NDCuboidMapper();
CuboidReducer reducer = new CuboidReducer();
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/kylin-it/src/test/java/org/apache/kylin/job/BaseTestDistributedScheduler.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/job/BaseTestDistributedScheduler.java b/kylin-it/src/test/java/org/apache/kylin/job/BaseTestDistributedScheduler.java
index aa96e2e..f410f59 100644
--- a/kylin-it/src/test/java/org/apache/kylin/job/BaseTestDistributedScheduler.java
+++ b/kylin-it/src/test/java/org/apache/kylin/job/BaseTestDistributedScheduler.java
@@ -76,7 +76,7 @@ public class BaseTestDistributedScheduler extends HBaseMetadataTestCase {
KylinConfig srcConfig = KylinConfig.getInstanceFromEnv();
localMetaDir = Files.createTempDir();
- backup = srcConfig.getMetadataUrl();
+ backup = srcConfig.getMetadataUrl().toString();
srcConfig.setProperty("kylin.metadata.url", localMetaDir.getAbsolutePath());
srcConfig.writeProperties(new File(confDstPath1));
srcConfig.writeProperties(new File(confDstPath2));
@@ -198,6 +198,7 @@ public class BaseTestDistributedScheduler extends HBaseMetadataTestCase {
}
private String getLockPath(String pathName) {
- return DistributedScheduler.ZOOKEEPER_LOCK_PATH + "/" + kylinConfig1.getMetadataUrlPrefix() + "/" + pathName;
+ return DistributedScheduler.dropDoubleSlash(//
+ DistributedScheduler.ZOOKEEPER_LOCK_PATH + "/" + kylinConfig1.getMetadataUrlPrefix() + "/" + pathName);
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
----------------------------------------------------------------------
diff --git a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
index 733ca06..847baf8 100644
--- a/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
+++ b/kylin-it/src/test/java/org/apache/kylin/storage/hbase/ITStorageTest.java
@@ -71,9 +71,8 @@ public class ITStorageTest extends HBaseMetadataTestCase {
cube = cubeMgr.getCube("test_kylin_cube_without_slr_left_join_empty");
Assert.assertNotNull(cube);
storageEngine = StorageFactory.createQuery(cube);
- String url = KylinConfig.getInstanceFromEnv().getStorageUrl();
context = new StorageContext();
- context.setConnUrl(url);
+ context.setConnUrl(KylinConfig.getInstanceFromEnv().getStorageUrl());
mockup = new StorageMockUtils(cube.getModel());
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/query/src/main/java/org/apache/kylin/query/schema/OLAPSchema.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/schema/OLAPSchema.java b/query/src/main/java/org/apache/kylin/query/schema/OLAPSchema.java
index 52023e4..9124b15 100644
--- a/query/src/main/java/org/apache/kylin/query/schema/OLAPSchema.java
+++ b/query/src/main/java/org/apache/kylin/query/schema/OLAPSchema.java
@@ -25,6 +25,7 @@ import java.util.Set;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.model.TableDesc;
@@ -40,7 +41,7 @@ public class OLAPSchema extends AbstractSchema {
private KylinConfig config;
private String projectName;
private String schemaName;
- private String storageUrl;
+ private StorageURL storageUrl;
private String starSchemaUrl;
private String starSchemaUser;
private String starSchemaPassword;
@@ -89,7 +90,7 @@ public class OLAPSchema extends AbstractSchema {
return schemaName;
}
- public String getStorageUrl() {
+ public StorageURL getStorageUrl() {
return storageUrl;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/server-base/src/main/java/org/apache/kylin/rest/security/MockAclHBaseStorage.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/security/MockAclHBaseStorage.java b/server-base/src/main/java/org/apache/kylin/rest/security/MockAclHBaseStorage.java
index 72d44bb..ca49641 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/security/MockAclHBaseStorage.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/security/MockAclHBaseStorage.java
@@ -18,14 +18,15 @@
package org.apache.kylin.rest.security;
+import java.io.IOException;
+
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Table;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.rest.service.LegacyAclService;
-import org.apache.kylin.rest.service.QueryService;
import org.apache.kylin.rest.service.LegacyUserService;
-
-import java.io.IOException;
+import org.apache.kylin.rest.service.QueryService;
/**
*/
@@ -40,8 +41,8 @@ public class MockAclHBaseStorage implements AclHBaseStorage {
private RealAclHBaseStorage realAcl;
public MockAclHBaseStorage() {
- String metadataUrl = KylinConfig.getInstanceFromEnv().getMetadataUrl();
- if (metadataUrl != null && metadataUrl.endsWith("hbase")) {
+ StorageURL metadataUrl = KylinConfig.getInstanceFromEnv().getMetadataUrl();
+ if (metadataUrl.getScheme().endsWith("hbase")) {
// hbase must be available since metadata is on it
// in this case, let us use a real ACL instead of mockup
realAcl = new RealAclHBaseStorage();
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/server-base/src/main/java/org/apache/kylin/rest/security/RealAclHBaseStorage.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/security/RealAclHBaseStorage.java b/server-base/src/main/java/org/apache/kylin/rest/security/RealAclHBaseStorage.java
index f42b3f2..98cef3a 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/security/RealAclHBaseStorage.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/security/RealAclHBaseStorage.java
@@ -18,33 +18,32 @@
package org.apache.kylin.rest.security;
+import java.io.IOException;
+
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Table;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.rest.service.LegacyAclService;
-import org.apache.kylin.rest.service.QueryService;
import org.apache.kylin.rest.service.LegacyUserService;
+import org.apache.kylin.rest.service.QueryService;
import org.apache.kylin.storage.hbase.HBaseConnection;
-import java.io.IOException;
-
/**
*/
@Deprecated
public class RealAclHBaseStorage implements AclHBaseStorage {
- private String hbaseUrl;
+ private StorageURL hbaseUrl;
private String aclTableName;
private String userTableName;
@Override
public String prepareHBaseTable(Class<?> clazz) throws IOException {
KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
- String metadataUrl = kylinConfig.getMetadataUrl();
- int cut = metadataUrl.indexOf('@');
- hbaseUrl = cut < 0 ? metadataUrl : metadataUrl.substring(cut + 1);
- String tableNameBase = kylinConfig.getMetadataUrlPrefix();
+ hbaseUrl = kylinConfig.getMetadataUrl();
+ String tableNameBase = hbaseUrl.getIdentifier();
if (clazz == LegacyAclService.class) {
aclTableName = tableNameBase + ACL_TABLE_NAME;
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
----------------------------------------------------------------------
diff --git a/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
index 1ddf4a3..9eab5a3 100644
--- a/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
+++ b/server-base/src/main/java/org/apache/kylin/rest/service/QueryService.java
@@ -55,6 +55,7 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.QueryContext;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.common.exceptions.ResourceLimitExceededException;
import org.apache.kylin.common.util.Bytes;
@@ -115,7 +116,7 @@ public class QueryService extends BasicService {
private final Serializer<Query[]> querySerializer = new Serializer<Query[]>(Query[].class);
private final BadQueryDetector badQueryDetector = new BadQueryDetector();
- private final String hbaseUrl;
+ private final StorageURL hbaseUrl;
private final String userTableName;
@Autowired
@@ -134,12 +135,8 @@ public class QueryService extends BasicService {
public QueryService() {
KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
- String metadataUrl = kylinConfig.getMetadataUrl();
- // split TABLE@HBASE_URL
- int cut = metadataUrl.indexOf('@');
- hbaseUrl = cut < 0 ? metadataUrl : metadataUrl.substring(cut + 1);
- String tableNameBase = kylinConfig.getMetadataUrlPrefix();
- userTableName = tableNameBase + USER_TABLE_NAME;
+ hbaseUrl = kylinConfig.getMetadataUrl();
+ userTableName = hbaseUrl.getIdentifier() + USER_TABLE_NAME;
badQueryDetector.start();
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseConnection.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseConnection.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseConnection.java
index f0b9428..2bbc2bc 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseConnection.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseConnection.java
@@ -18,7 +18,19 @@
package org.apache.kylin.storage.hbase;
-import com.google.common.collect.Sets;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -33,22 +45,13 @@ import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.persistence.StorageException;
import org.apache.kylin.common.util.HadoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import com.google.common.collect.Sets;
/**
* @author yangli9
@@ -60,8 +63,8 @@ public class HBaseConnection {
private static final Logger logger = LoggerFactory.getLogger(HBaseConnection.class);
- private static final Map<String, Configuration> configCache = new ConcurrentHashMap<String, Configuration>();
- private static final Map<String, Connection> connPool = new ConcurrentHashMap<String, Connection>();
+ private static final Map<StorageURL, Configuration> configCache = new ConcurrentHashMap<StorageURL, Configuration>();
+ private static final Map<StorageURL, Connection> connPool = new ConcurrentHashMap<StorageURL, Connection>();
private static final ThreadLocal<Configuration> configThreadLocal = new ThreadLocal<>();
private static ExecutorService coprocessorPool = null;
@@ -76,7 +79,7 @@ public class HBaseConnection {
try {
conn.close();
} catch (IOException e) {
- e.printStackTrace();
+ logger.error("error closing hbase connection " + conn, e);
}
}
}
@@ -131,15 +134,15 @@ public class HBaseConnection {
public static Configuration getCurrentHBaseConfiguration() {
if (configThreadLocal.get() == null) {
- String storageUrl = KylinConfig.getInstanceFromEnv().getStorageUrl();
+ StorageURL storageUrl = KylinConfig.getInstanceFromEnv().getStorageUrl();
configThreadLocal.set(newHBaseConfiguration(storageUrl));
}
return configThreadLocal.get();
}
- private static Configuration newHBaseConfiguration(String url) {
+ private static Configuration newHBaseConfiguration(StorageURL url) {
// using a hbase:xxx URL is deprecated, instead hbase config is always loaded from hbase-site.xml in classpath
- if (!(StringUtils.isEmpty(url) || "hbase".equals(url)))
+ if (!"hbase".equals(url.getScheme()))
throw new IllegalArgumentException("to use hbase storage, pls set 'kylin.storage.url=hbase' in kylin.properties");
Configuration conf = HBaseConfiguration.create(HadoopUtil.getCurrentConfiguration());
@@ -159,6 +162,10 @@ public class HBaseConnection {
if (StringUtils.isBlank(conf.get("hbase.fs.tmp.dir"))) {
conf.set("hbase.fs.tmp.dir", "/tmp");
}
+
+ for (Entry<String, String> entry : url.getAllParameters().entrySet()) {
+ conf.set(entry.getKey(), entry.getValue());
+ }
return conf;
}
@@ -210,7 +217,7 @@ public class HBaseConnection {
// returned Connection can be shared by multiple threads and does not require close()
@SuppressWarnings("resource")
- public static Connection get(String url) {
+ public static Connection get(StorageURL url) {
// find configuration
Configuration conf = configCache.get(url);
if (conf == null) {
@@ -252,15 +259,15 @@ public class HBaseConnection {
}
}
- public static boolean tableExists(String hbaseUrl, String tableName) throws IOException {
+ public static boolean tableExists(StorageURL hbaseUrl, String tableName) throws IOException {
return tableExists(HBaseConnection.get(hbaseUrl), tableName);
}
- public static void createHTableIfNeeded(String hbaseUrl, String tableName, String... families) throws IOException {
+ public static void createHTableIfNeeded(StorageURL hbaseUrl, String tableName, String... families) throws IOException {
createHTableIfNeeded(HBaseConnection.get(hbaseUrl), tableName, families);
}
- public static void deleteTable(String hbaseUrl, String tableName) throws IOException {
+ public static void deleteTable(StorageURL hbaseUrl, String tableName) throws IOException {
deleteTable(HBaseConnection.get(hbaseUrl), tableName);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
index 76aefe0..a2e0229 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/HBaseResourceStore.java
@@ -22,7 +22,9 @@ import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.NavigableSet;
import java.util.TreeSet;
import java.util.UUID;
@@ -31,13 +33,12 @@ import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
@@ -49,6 +50,7 @@ import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.persistence.RawResource;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.StringEntity;
@@ -71,25 +73,34 @@ public class HBaseResourceStore extends ResourceStore {
private static final String COLUMN_TS = "t";
private static final byte[] B_COLUMN_TS = Bytes.toBytes(COLUMN_TS);
- final String tableNameBase;
- final String hbaseUrl;
+ final String tableName;
+ final StorageURL metadataUrl;
Connection getConnection() throws IOException {
- return HBaseConnection.get(hbaseUrl);
+ return HBaseConnection.get(metadataUrl);
}
public HBaseResourceStore(KylinConfig kylinConfig) throws IOException {
super(kylinConfig);
- String metadataUrl = kylinConfig.getMetadataUrl();
- // split TABLE@HBASE_URL
- int cut = metadataUrl.indexOf('@');
- tableNameBase = cut < 0 ? DEFAULT_STORE_NAME : metadataUrl.substring(0, cut);
- hbaseUrl = cut < 0 ? metadataUrl : metadataUrl.substring(cut + 1);
- if (!hbaseUrl.equals("hbase"))
- throw new IOException("Can not create HBaseResourceStore. Url not match. Url:" + hbaseUrl);
+ metadataUrl = buildMetadataUrl(kylinConfig);
+ tableName = metadataUrl.getIdentifier();
createHTableIfNeeded(getAllInOneTableName());
+ }
+
+ private StorageURL buildMetadataUrl(KylinConfig kylinConfig) throws IOException {
+ StorageURL url = kylinConfig.getMetadataUrl();
+ if (!url.getScheme().equals("hbase"))
+ throw new IOException("Cannot create HBaseResourceStore. Url not match. Url: " + url);
+ // control timeout for prompt error report
+ Map<String, String> newParams = new LinkedHashMap<>();
+ newParams.put("hbase.client.scanner.timeout.period", "10000");
+ newParams.put("hbase.rpc.timeout", "5000");
+ newParams.put("hbase.client.retries.number", "1");
+ newParams.putAll(url.getAllParameters());
+
+ return url.copy(newParams);
}
private void createHTableIfNeeded(String tableName) throws IOException {
@@ -97,7 +108,7 @@ public class HBaseResourceStore extends ResourceStore {
}
private String getAllInOneTableName() {
- return tableNameBase;
+ return tableName;
}
@Override
@@ -120,14 +131,11 @@ public class HBaseResourceStore extends ResourceStore {
return result.isEmpty() ? null : result;
}
- /*
- override get meta store uuid method for backward compatibility
- */
-
+ /* override get meta store uuid method for backward compatibility */
@Override
public String createMetaStoreUUID() throws IOException {
- try (final HBaseAdmin hbaseAdmin = new HBaseAdmin(HBaseConfiguration.create(HadoopUtil.getCurrentConfiguration()))) {
- final String metaStoreName = KylinConfig.getInstanceFromEnv().getMetadataUrlPrefix();
+ try (final Admin hbaseAdmin = HBaseConnection.get(metadataUrl).getAdmin()) {
+ final String metaStoreName = metadataUrl.getIdentifier();
final HTableDescriptor desc = hbaseAdmin.getTableDescriptor(TableName.valueOf(metaStoreName));
String uuid = desc.getValue(HBaseConnection.HTABLE_UUID_TAG);
if (uuid != null)
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
index 581de38..2154ed1 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/CubeMigrationCLI.java
@@ -40,6 +40,7 @@ import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.persistence.JsonSerializer;
import org.apache.kylin.common.persistence.RawResource;
import org.apache.kylin.common.persistence.ResourceStore;
@@ -172,26 +173,15 @@ public class CubeMigrationCLI {
checkCLI.execute(cubeName);
}
- private static String checkAndGetHbaseUrl() {
- String srcMetadataUrl = srcConfig.getMetadataUrl();
- String dstMetadataUrl = dstConfig.getMetadataUrl();
+ private static void checkAndGetHbaseUrl() {
+ StorageURL srcMetadataUrl = srcConfig.getMetadataUrl();
+ StorageURL dstMetadataUrl = dstConfig.getMetadataUrl();
logger.info("src metadata url is " + srcMetadataUrl);
logger.info("dst metadata url is " + dstMetadataUrl);
- int srcIndex = srcMetadataUrl.toLowerCase().indexOf("hbase");
- int dstIndex = dstMetadataUrl.toLowerCase().indexOf("hbase");
- if (srcIndex < 0 || dstIndex < 0)
+ if (!"hbase".equals(srcMetadataUrl.getScheme()) || !"hbase".equals(dstMetadataUrl.getScheme()))
throw new IllegalStateException("Both metadata urls should be hbase metadata url");
-
- String srcHbaseUrl = srcMetadataUrl.substring(srcIndex).trim();
- String dstHbaseUrl = dstMetadataUrl.substring(dstIndex).trim();
- if (!srcHbaseUrl.equalsIgnoreCase(dstHbaseUrl)) {
- throw new IllegalStateException("hbase url not equal! ");
- }
-
- logger.info("hbase url is " + srcHbaseUrl.trim());
- return srcHbaseUrl.trim();
}
private static void renameFoldersInHdfs(CubeInstance cube) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/GridTableHBaseBenchmark.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/GridTableHBaseBenchmark.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/GridTableHBaseBenchmark.java
index dd5f8fa..a317110 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/GridTableHBaseBenchmark.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/util/GridTableHBaseBenchmark.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.storage.hbase.HBaseConnection;
@@ -73,7 +74,7 @@ public class GridTableHBaseBenchmark {
public static void testGridTable(double hitRatio, double indexRatio) throws IOException {
System.out.println("Testing grid table scanning, hit ratio " + hitRatio + ", index ratio " + indexRatio);
- String hbaseUrl = "hbase"; // use hbase-site.xml on classpath
+ StorageURL hbaseUrl = StorageURL.valueOf("default@hbase"); // use hbase-site.xml on classpath
Connection conn = HBaseConnection.get(hbaseUrl);
createHTableIfNeeded(conn, TEST_TABLE);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/main/java/org/apache/kylin/storage/hdfs/HDFSResourceStore.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hdfs/HDFSResourceStore.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hdfs/HDFSResourceStore.java
index b025cf7..fe1ad4e 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hdfs/HDFSResourceStore.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hdfs/HDFSResourceStore.java
@@ -18,7 +18,14 @@
package org.apache.kylin.storage.hdfs;
-import com.google.common.collect.Lists;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.TreeSet;
+
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -26,19 +33,14 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.persistence.RawResource;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.util.HadoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Collections;
-import java.util.List;
-import java.util.NavigableSet;
-import java.util.TreeSet;
+import com.google.common.collect.Lists;
public class HDFSResourceStore extends ResourceStore {
@@ -50,16 +52,12 @@ public class HDFSResourceStore extends ResourceStore {
public HDFSResourceStore(KylinConfig kylinConfig) throws Exception {
super(kylinConfig);
- String metadataUrl = kylinConfig.getMetadataUrl();
- int cut = metadataUrl.indexOf('@');
- if (cut < 0) {
- throw new IOException("kylin.metadata.url not recognized for HDFSResourceStore: " + metadataUrl);
- }
- String suffix = metadataUrl.substring(cut + 1);
- if (!suffix.equals("hdfs"))
- throw new IOException("kylin.metadata.url not recognized for HDFSResourceStore:" + suffix);
+ StorageURL metadataUrl = kylinConfig.getMetadataUrl();
+
+ if (!metadataUrl.getScheme().equals("hdfs"))
+ throw new IOException("kylin.metadata.url not recognized for HDFSResourceStore:" + metadataUrl);
- String path = metadataUrl.substring(0, cut);
+ String path = metadataUrl.getIdentifier();
fs = HadoopUtil.getFileSystem(path);
Path metadataPath = new Path(path);
if (fs.exists(metadataPath) == false) {
@@ -220,22 +218,4 @@ public class HDFSResourceStore extends ResourceStore {
resourcePath = resourcePath.substring(1, resourcePath.length());
return new Path(this.hdfsMetaPath, resourcePath);
}
-
- private static String getRelativePath(Path hdfsPath) {
- String path = hdfsPath.toString();
- int index = path.indexOf("://");
- if (index > 0) {
- path = path.substring(index + 3);
- }
-
- if (path.startsWith("/") == false) {
- if (path.indexOf("/") > 0) {
- path = path.substring(path.indexOf("/"));
- } else {
- path = "/" + path;
- }
- }
- return path;
- }
-
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
index f36a62c..d47f393 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
@@ -55,7 +55,7 @@ public class CubeHFileMapper2Test extends LocalFileMetadataTestCase {
this.createTestMetadata();
// hack for distributed cache
FileUtils.deleteDirectory(new File("../job/meta"));
- FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl()), new File("../job/meta"));
+ FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl().toString()), new File("../job/meta"));
CubeDesc desc = CubeManager.getInstance(getTestConfig()).getCube(cubeName).getDescriptor();
codec = new MeasureCodec(desc.getMeasures());
}
@@ -70,7 +70,7 @@ public class CubeHFileMapper2Test extends LocalFileMetadataTestCase {
public void testBasic() throws Exception {
Configuration hconf = HadoopUtil.getCurrentConfiguration();
- Context context = MockupMapContext.create(hconf, getTestConfig().getMetadataUrl(), cubeName, outKV);
+ Context context = MockupMapContext.create(hconf, cubeName, outKV);
CubeHFileMapper mapper = new CubeHFileMapper();
mapper.setup(context);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/MockupMapContext.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/MockupMapContext.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/MockupMapContext.java
index 536634d..61170e8 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/MockupMapContext.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/MockupMapContext.java
@@ -53,7 +53,7 @@ public class MockupMapContext implements MapContext {
private Object[] outKV;
- public static Context create(final Configuration hconf, String metadataUrl, String cubeName, final Object[] outKV) {
+ public static Context create(final Configuration hconf, String cubeName, final Object[] outKV) {
hconf.set(BatchConstants.CFG_CUBE_NAME, cubeName);
http://git-wip-us.apache.org/repos/asf/kylin/blob/2dfc5bb3/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
----------------------------------------------------------------------
diff --git a/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java b/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
index c162a76..08d4292 100644
--- a/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
+++ b/tool/src/main/java/org/apache/kylin/tool/CubeMigrationCLI.java
@@ -40,6 +40,7 @@ import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.StorageURL;
import org.apache.kylin.common.persistence.JsonSerializer;
import org.apache.kylin.common.persistence.RawResource;
import org.apache.kylin.common.persistence.ResourceStore;
@@ -172,26 +173,15 @@ public class CubeMigrationCLI {
checkCLI.execute(cubeName);
}
- private String checkAndGetHbaseUrl() {
- String srcMetadataUrl = srcConfig.getMetadataUrl();
- String dstMetadataUrl = dstConfig.getMetadataUrl();
+ private void checkAndGetHbaseUrl() {
+ StorageURL srcMetadataUrl = srcConfig.getMetadataUrl();
+ StorageURL dstMetadataUrl = dstConfig.getMetadataUrl();
logger.info("src metadata url is " + srcMetadataUrl);
logger.info("dst metadata url is " + dstMetadataUrl);
- int srcIndex = srcMetadataUrl.toLowerCase().indexOf("hbase");
- int dstIndex = dstMetadataUrl.toLowerCase().indexOf("hbase");
- if (srcIndex < 0 || dstIndex < 0)
+ if (!"hbase".equals(srcMetadataUrl.getScheme()) || !"hbase".equals(dstMetadataUrl.getScheme()))
throw new IllegalStateException("Both metadata urls should be hbase metadata url");
-
- String srcHbaseUrl = srcMetadataUrl.substring(srcIndex).trim();
- String dstHbaseUrl = dstMetadataUrl.substring(dstIndex).trim();
- if (!srcHbaseUrl.equalsIgnoreCase(dstHbaseUrl)) {
- throw new IllegalStateException("hbase url not equal! ");
- }
-
- logger.info("hbase url is " + srcHbaseUrl.trim());
- return srcHbaseUrl.trim();
}
protected void renameFoldersInHdfs(CubeInstance cube) throws IOException {