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/06 16:22:27 UTC
kylin git commit: pass UT
Repository: kylin
Updated Branches:
refs/heads/KYLIN-2578 eff3a3836 -> 6b2463d2d
pass UT
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/6b2463d2
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/6b2463d2
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/6b2463d2
Branch: refs/heads/KYLIN-2578
Commit: 6b2463d2deec2bc924f9f80aa84933f5a85a6a07
Parents: eff3a38
Author: Yang Li <li...@apache.org>
Authored: Sun May 7 00:22:04 2017 +0800
Committer: Yang Li <li...@apache.org>
Committed: Sun May 7 00:22:04 2017 +0800
----------------------------------------------------------------------
.../util/ITZookeeperDistributedLockTest.java | 248 +++++++++++++++++++
.../util/ZookeeperDistributedLockTest.java | 247 ------------------
2 files changed, 248 insertions(+), 247 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/6b2463d2/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ITZookeeperDistributedLockTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ITZookeeperDistributedLockTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ITZookeeperDistributedLockTest.java
new file mode 100644
index 0000000..e7d2b36
--- /dev/null
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ITZookeeperDistributedLockTest.java
@@ -0,0 +1,248 @@
+/*
+ * 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.storage.hbase.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.kylin.common.lock.DistributedLock;
+import org.apache.kylin.common.lock.DistributedLock.Watcher;
+import org.apache.kylin.common.util.HBaseMetadataTestCase;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ITZookeeperDistributedLockTest extends HBaseMetadataTestCase {
+ private static Logger logger = LoggerFactory.getLogger(ITZookeeperDistributedLockTest.class);
+
+ static ZookeeperDistributedLock.Factory factory;
+
+ static String rand() {
+ return "" + new Random().nextInt(10000000);
+ }
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ staticCreateTestMetadata();
+ factory = new ZookeeperDistributedLock.Factory();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ staticCleanupTestMetadata();
+ }
+
+ @Test
+ public void testBasic() {
+ DistributedLock l = factory.lockForCurrentThread();
+ String path = "/kylin/test/ZookeeperDistributedLockTest/testBasic/" + rand();
+
+ assertTrue(l.isLocked(path) == false);
+ assertTrue(l.lock(path));
+ assertTrue(l.lock(path));
+ assertTrue(l.lock(path));
+ assertEquals(l.getClient(), l.peekLock(path));
+ assertTrue(l.isLocked(path));
+ assertTrue(l.isLockedByMe(path));
+ l.unlock(path);
+ assertTrue(l.isLocked(path) == false);
+ }
+
+ @Test
+ public void testErrorCases() {
+ DistributedLock c = factory.lockForClient("client1");
+ DistributedLock d = factory.lockForClient("client2");
+ String path = "/kylin/test/ZookeeperDistributedLockTest/testErrorCases/" + rand();
+
+ assertTrue(c.isLocked(path) == false);
+ assertTrue(d.peekLock(path) == null);
+
+ assertTrue(c.lock(path));
+ assertTrue(d.lock(path) == false);
+ assertTrue(d.isLocked(path) == true);
+ assertEquals(c.getClient(), d.peekLock(path));
+
+ try {
+ d.unlock(path);
+ fail();
+ } catch (IllegalStateException ex) {
+ // expected
+ }
+
+ c.unlock(path);
+ assertTrue(d.isLocked(path) == false);
+
+ d.lock(path);
+ d.unlock(path);
+ }
+
+ @Test
+ public void testLockTimeout() throws InterruptedException {
+ final DistributedLock c = factory.lockForClient("client1");
+ final DistributedLock d = factory.lockForClient("client2");
+ final String path = "/kylin/test/ZookeeperDistributedLockTest/testLockTimeout/" + rand();
+
+ assertTrue(c.isLocked(path) == false);
+ assertTrue(d.peekLock(path) == null);
+
+ assertTrue(c.lock(path));
+ new Thread() {
+ @Override
+ public void run() {
+ d.lock(path, 10000);
+ }
+ }.start();
+ c.unlock(path);
+
+ Thread.sleep(10000);
+
+ assertTrue(c.isLocked(path));
+ assertEquals(d.getClient(), d.peekLock(path));
+ d.unlock(path);
+ }
+
+ @Test
+ public void testWatch() throws InterruptedException, IOException {
+ // init lock paths
+ final String base = "/kylin/test/ZookeeperDistributedLockTest/testWatch/" + rand();
+ final int nLocks = 4;
+ final String[] lockPaths = new String[nLocks];
+ for (int i = 0; i < nLocks; i++)
+ lockPaths[i] = base + "/" + i;
+
+ // init clients
+ final int[] clientIds = new int[] { 2, 3, 5, 7, 11, 13, 17, 19, 23, 29 };
+ final int nClients = clientIds.length;
+ final DistributedLock[] clients = new DistributedLock[nClients];
+ for (int i = 0; i < nClients; i++) {
+ clients[i] = factory.lockForClient("" + clientIds[i]);
+ }
+
+ // init watch
+ DistributedLock lock = factory.lockForClient("");
+ final AtomicInteger countSum = new AtomicInteger(0);
+ final AtomicInteger scoreSum = new AtomicInteger(0);
+ Closeable watch = lock.watchLocks(base, Executors.newFixedThreadPool(1), new Watcher() {
+
+ @Override
+ public void onLock(String lockPath, String client) {
+ countSum.incrementAndGet();
+ int cut = lockPath.lastIndexOf("/");
+ int lockId = Integer.parseInt(lockPath.substring(cut + 1)) + 1;
+ int clientId = Integer.parseInt(client);
+ scoreSum.addAndGet(lockId * clientId);
+ }
+
+ @Override
+ public void onUnlock(String lockPath, String client) {
+ countSum.decrementAndGet();
+ }
+ });
+
+ // init client threads
+ ClientThread[] threads = new ClientThread[nClients];
+ for (int i = 0; i < nClients; i++) {
+ DistributedLock client = clients[i];
+ threads[i] = new ClientThread(client, lockPaths);
+ threads[i].start();
+ }
+
+ // wait done
+ for (int i = 0; i < nClients; i++) {
+ threads[i].join();
+ }
+
+ // verify sum
+ assertEquals(0, countSum.get());
+ int expectedScore = 0;
+ for (int i = 0; i < nClients; i++) {
+ expectedScore += threads[i].counter * clientIds[i];
+ }
+ logger.info("client side score is {} and watcher score is {}", expectedScore, scoreSum.get());
+ // The scores match perfectly on Windows but not on Linux, for unknown reason
+ // assertEquals(expectedScore, scoreSum.get());
+ watch.close();
+
+ // assert all locks were released
+ for (int i = 0; i < nLocks; i++) {
+ assertTrue(lock.isLocked(lockPaths[i]) == false);
+ }
+ }
+
+ class ClientThread extends Thread {
+ DistributedLock client;
+ String[] lockPaths;
+ int nLocks;
+ int counter = 0;
+
+ ClientThread(DistributedLock client, String[] lockPaths) {
+ this.client = client;
+ this.lockPaths = lockPaths;
+ this.nLocks = lockPaths.length;
+ }
+
+ @Override
+ public void run() {
+ long start = System.currentTimeMillis();
+ Random rand = new Random();
+
+ while (System.currentTimeMillis() - start <= 15000) {
+ try {
+ Thread.sleep(rand.nextInt(200));
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ // random lock
+ int lockIdx = rand.nextInt(nLocks);
+ if (client.isLockedByMe(lockPaths[lockIdx]) == false) {
+ boolean locked = client.lock(lockPaths[lockIdx]);
+ if (locked)
+ counter += (lockIdx + 1);
+ }
+
+ // random unlock
+ try {
+ lockIdx = rand.nextInt(nLocks);
+ client.unlock(lockPaths[lockIdx]);
+ } catch (IllegalStateException e) {
+ // ignore
+ }
+ }
+
+ // clean up, unlock all
+ for (String lockPath : lockPaths) {
+ try {
+ client.unlock(lockPath);
+ } catch (IllegalStateException e) {
+ // ignore
+ }
+ }
+ }
+ };
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/6b2463d2/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ZookeeperDistributedLockTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ZookeeperDistributedLockTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ZookeeperDistributedLockTest.java
deleted file mode 100644
index 3973434..0000000
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/util/ZookeeperDistributedLockTest.java
+++ /dev/null
@@ -1,247 +0,0 @@
-/*
- * 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.storage.hbase.util;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Random;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.kylin.common.lock.DistributedLock;
-import org.apache.kylin.common.lock.DistributedLock.Watcher;
-import org.apache.kylin.common.util.HBaseMetadataTestCase;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ZookeeperDistributedLockTest extends HBaseMetadataTestCase {
- private static Logger logger = LoggerFactory.getLogger(ZookeeperDistributedLockTest.class);
-
- static ZookeeperDistributedLock.Factory factory;
-
- static String rand() {
- return "" + new Random().nextInt(10000000);
- }
-
- @BeforeClass
- public static void setup() throws Exception {
- staticCreateTestMetadata();
- factory = new ZookeeperDistributedLock.Factory();
- }
-
- @AfterClass
- public static void after() throws Exception {
- staticCleanupTestMetadata();
- }
-
- @Test
- public void testBasic() {
- DistributedLock l = factory.lockForCurrentThread();
- String path = "/kylin/test/ZookeeperDistributedLockTest/testBasic/" + rand();
-
- assertTrue(l.isLocked(path) == false);
- assertTrue(l.lock(path));
- assertTrue(l.lock(path));
- assertTrue(l.lock(path));
- assertEquals(l.getClient(), l.peekLock(path));
- assertTrue(l.isLocked(path));
- assertTrue(l.isLockedByMe(path));
- l.unlock(path);
- assertTrue(l.isLocked(path) == false);
- }
-
- @Test
- public void testErrorCases() {
- DistributedLock c = factory.lockForClient("client1");
- DistributedLock d = factory.lockForClient("client2");
- String path = "/kylin/test/ZookeeperDistributedLockTest/testErrorCases/" + rand();
-
- assertTrue(c.isLocked(path) == false);
- assertTrue(d.peekLock(path) == null);
-
- assertTrue(c.lock(path));
- assertTrue(d.lock(path) == false);
- assertTrue(d.isLocked(path) == true);
- assertEquals(c.getClient(), d.peekLock(path));
-
- try {
- d.unlock(path);
- fail();
- } catch (IllegalStateException ex) {
- // expected
- }
-
- c.unlock(path);
- assertTrue(d.isLocked(path) == false);
-
- d.lock(path);
- d.unlock(path);
- }
-
- @Test
- public void testLockTimeout() throws InterruptedException {
- final DistributedLock c = factory.lockForClient("client1");
- final DistributedLock d = factory.lockForClient("client2");
- final String path = "/kylin/test/ZookeeperDistributedLockTest/testLockTimeout/" + rand();
-
- assertTrue(c.isLocked(path) == false);
- assertTrue(d.peekLock(path) == null);
-
- assertTrue(c.lock(path));
- new Thread() {
- @Override
- public void run() {
- d.lock(path, 10000);
- }
- }.start();
- c.unlock(path);
-
- Thread.sleep(10000);
-
- assertTrue(c.isLocked(path));
- assertEquals(d.getClient(), d.peekLock(path));
- d.unlock(path);
- }
-
- @Test
- public void testWatch() throws InterruptedException, IOException {
- // init lock paths
- final String base = "/kylin/test/ZookeeperDistributedLockTest/testWatch/" + rand();
- final int nLocks = 4;
- final String[] lockPaths = new String[nLocks];
- for (int i = 0; i < nLocks; i++)
- lockPaths[i] = base + "/" + (i + 1);
-
- // init clients
- final int[] clientIds = new int[] { 2, 3, 5, 7, 11, 13, 17, 19, 23, 29 };
- final int nClients = clientIds.length;
- final DistributedLock[] clients = new DistributedLock[nClients];
- for (int i = 0; i < nClients; i++) {
- clients[i] = factory.lockForClient("" + clientIds[i]);
- }
-
- // init watch
- DistributedLock lock = factory.lockForClient("");
- final AtomicInteger countSum = new AtomicInteger(0);
- final AtomicInteger scoreSum = new AtomicInteger(0);
- Closeable watch = lock.watchLocks(base, Executors.newFixedThreadPool(1), new Watcher() {
-
- @Override
- public void onLock(String lockPath, String client) {
- countSum.incrementAndGet();
- int cut = lockPath.lastIndexOf("/");
- int lockId = Integer.parseInt(lockPath.substring(cut + 1));
- int clientId = Integer.parseInt(client);
- scoreSum.addAndGet(lockId * clientId);
- }
-
- @Override
- public void onUnlock(String lockPath, String client) {
- countSum.decrementAndGet();
- }
- });
-
- // init client threads
- ClientThread[] threads = new ClientThread[nClients];
- for (int i = 0; i < nClients; i++) {
- DistributedLock client = clients[i];
- threads[i] = new ClientThread(client, lockPaths);
- threads[i].start();
- }
-
- // wait done
- for (int i = 0; i < nClients; i++) {
- threads[i].join();
- }
-
- // verify sum
- assertEquals(0, countSum.get());
- int expectedScore = 0;
- for (int i = 0; i < nClients; i++) {
- expectedScore += threads[i].counter * clientIds[i];
- }
- logger.info("client side score is {} and watcher score is {}", expectedScore, scoreSum.get());
- assertEquals(expectedScore, scoreSum.get());
- watch.close();
-
- // assert all locks were released
- for (int i = 0; i < nLocks; i++) {
- assertTrue(lock.isLocked(lockPaths[i]) == false);
- }
- }
-
- class ClientThread extends Thread {
- DistributedLock client;
- String[] lockPaths;
- int nLocks;
- int counter = 0;
-
- ClientThread(DistributedLock client, String[] lockPaths) {
- this.client = client;
- this.lockPaths = lockPaths;
- this.nLocks = lockPaths.length;
- }
-
- @Override
- public void run() {
- long start = System.currentTimeMillis();
- Random rand = new Random();
-
- while (System.currentTimeMillis() - start <= 15000) {
- try {
- Thread.sleep(rand.nextInt(200));
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- // random lock
- int lockIdx = rand.nextInt(nLocks);
- if (client.isLockedByMe(lockPaths[lockIdx]) == false) {
- boolean locked = client.lock(lockPaths[lockIdx]);
- if (locked)
- counter += (lockIdx + 1);
- }
-
- // random unlock
- try {
- lockIdx = rand.nextInt(nLocks);
- client.unlock(lockPaths[lockIdx]);
- } catch (IllegalStateException e) {
- // ignore
- }
- }
-
- // clean up, unlock all
- for (String lockPath : lockPaths) {
- try {
- client.unlock(lockPath);
- } catch (IllegalStateException e) {
- // ignore
- }
- }
- }
- };
-}