You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ap...@apache.org on 2018/06/02 00:00:59 UTC
[1/2] hbase git commit: HBASE-20667 Rename TestGlobalThrottler to
TestReplicationGlobalThrottler
Repository: hbase
Updated Branches:
refs/heads/branch-1 a25878cf4 -> 07487cb0b
refs/heads/branch-2.0 512397220 -> 20c16ae39
HBASE-20667 Rename TestGlobalThrottler to TestReplicationGlobalThrottler
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/20c16ae3
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/20c16ae3
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/20c16ae3
Branch: refs/heads/branch-2.0
Commit: 20c16ae3911152d2813936c2760c36047aa29271
Parents: 5123972
Author: Andrew Purtell <ap...@apache.org>
Authored: Fri Jun 1 16:44:11 2018 -0700
Committer: Andrew Purtell <ap...@apache.org>
Committed: Fri Jun 1 16:46:07 2018 -0700
----------------------------------------------------------------------
.../TestGlobalReplicationThrottler.java | 190 ++++++++++++++++++
.../regionserver/TestGlobalThrottler.java | 193 -------------------
2 files changed, 190 insertions(+), 193 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/20c16ae3/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java
new file mode 100644
index 0000000..13bb58a
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java
@@ -0,0 +1,190 @@
+/**
+ * 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.hadoop.hbase.replication.regionserver;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.HTestConst;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+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.client.replication.ReplicationAdmin;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
+import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Category({ ReplicationTests.class, LargeTests.class })
+public class TestGlobalReplicationThrottler {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestGlobalReplicationThrottler.class);
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestGlobalReplicationThrottler.class);
+ private static Configuration conf1;
+ private static Configuration conf2;
+
+ private static HBaseTestingUtility utility1;
+ private static HBaseTestingUtility utility2;
+
+ private static final byte[] famName = Bytes.toBytes("f");
+ private static final byte[] VALUE = Bytes.toBytes("v");
+ private static final byte[] ROW = Bytes.toBytes("r");
+ private static final byte[][] ROWS = HTestConst.makeNAscii(ROW, 100);
+
+ @Rule
+ public TestName name = new TestName();
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ conf1 = HBaseConfiguration.create();
+ conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
+ conf1.setLong("replication.source.sleepforretries", 100);
+ // Each WAL is about 120 bytes
+ conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, 200);
+ conf1.setLong("replication.source.per.peer.node.bandwidth", 100L);
+
+ utility1 = new HBaseTestingUtility(conf1);
+ utility1.startMiniZKCluster();
+ MiniZooKeeperCluster miniZK = utility1.getZkCluster();
+ new ZKWatcher(conf1, "cluster1", null, true);
+
+ conf2 = new Configuration(conf1);
+ conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
+
+ utility2 = new HBaseTestingUtility(conf2);
+ utility2.setZkCluster(miniZK);
+ new ZKWatcher(conf2, "cluster2", null, true);
+
+ ReplicationAdmin admin1 = new ReplicationAdmin(conf1);
+ ReplicationPeerConfig rpc = new ReplicationPeerConfig();
+ rpc.setClusterKey(utility2.getClusterKey());
+
+ utility1.startMiniCluster(1, 1);
+ utility2.startMiniCluster(1, 1);
+
+ admin1.addPeer("peer1", rpc, null);
+ admin1.addPeer("peer2", rpc, null);
+ admin1.addPeer("peer3", rpc, null);
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ utility2.shutdownMiniCluster();
+ utility1.shutdownMiniCluster();
+ }
+
+
+ volatile private boolean testQuotaPass = false;
+ volatile private boolean testQuotaNonZero = false;
+ @Test
+ public void testQuota() throws IOException {
+ final TableName tableName = TableName.valueOf(name.getMethodName());
+ HTableDescriptor table = new HTableDescriptor(tableName);
+ HColumnDescriptor fam = new HColumnDescriptor(famName);
+ fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
+ table.addFamily(fam);
+ utility1.getAdmin().createTable(table);
+ utility2.getAdmin().createTable(table);
+
+ Thread watcher = new Thread(()->{
+ Replication replication = (Replication)utility1.getMiniHBaseCluster()
+ .getRegionServer(0).getReplicationSourceService();
+ AtomicLong bufferUsed = replication.getReplicationManager().getTotalBufferUsed();
+ testQuotaPass = true;
+ while (!Thread.interrupted()) {
+ long size = bufferUsed.get();
+ if (size > 0) {
+ testQuotaNonZero = true;
+ }
+ if (size > 600) {
+ // We read logs first then check throttler, so if the buffer quota limiter doesn't
+ // take effect, it will push many logs and exceed the quota.
+ testQuotaPass = false;
+ }
+ Threads.sleep(50);
+ }
+ });
+ watcher.start();
+
+ try(Table t1 = utility1.getConnection().getTable(tableName);
+ Table t2 = utility2.getConnection().getTable(tableName)) {
+ for (int i = 0; i < 50; i++) {
+ Put put = new Put(ROWS[i]);
+ put.addColumn(famName, VALUE, VALUE);
+ t1.put(put);
+ }
+ long start = EnvironmentEdgeManager.currentTime();
+ while (EnvironmentEdgeManager.currentTime() - start < 180000) {
+ Scan scan = new Scan();
+ scan.setCaching(50);
+ int count = 0;
+ try (ResultScanner results = t2.getScanner(scan)) {
+ for (Result result : results) {
+ count++;
+ }
+ }
+ if (count < 50) {
+ LOG.info("Waiting all logs pushed to slave. Expected 50 , actual " + count);
+ Threads.sleep(200);
+ continue;
+ }
+ break;
+ }
+ }
+
+ watcher.interrupt();
+ Assert.assertTrue(testQuotaPass);
+ Assert.assertTrue(testQuotaNonZero);
+ }
+
+ private List<Integer> getRowNumbers(List<Cell> cells) {
+ List<Integer> listOfRowNumbers = new ArrayList<>(cells.size());
+ for (Cell c : cells) {
+ listOfRowNumbers.add(Integer.parseInt(Bytes
+ .toString(c.getRowArray(), c.getRowOffset() + ROW.length,
+ c.getRowLength() - ROW.length)));
+ }
+ return listOfRowNumbers;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/20c16ae3/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
deleted file mode 100644
index d3b4e8e..0000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
+++ /dev/null
@@ -1,193 +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.hadoop.hbase.replication.regionserver;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.HBaseClassTestRule;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.HTestConst;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-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.client.replication.ReplicationAdmin;
-import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
-import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.apache.hadoop.hbase.testclassification.ReplicationTests;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.hadoop.hbase.util.Threads;
-import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
-import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.rules.TestName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Category({ ReplicationTests.class, LargeTests.class })
-public class TestGlobalThrottler {
-
- @ClassRule
- public static final HBaseClassTestRule CLASS_RULE =
- HBaseClassTestRule.forClass(TestGlobalThrottler.class);
-
- private static final Logger LOG = LoggerFactory.getLogger(TestGlobalThrottler.class);
- private static Configuration conf1;
- private static Configuration conf2;
-
- private static HBaseTestingUtility utility1;
- private static HBaseTestingUtility utility2;
-
- private static final byte[] famName = Bytes.toBytes("f");
- private static final byte[] VALUE = Bytes.toBytes("v");
- private static final byte[] ROW = Bytes.toBytes("r");
- private static final byte[][] ROWS = HTestConst.makeNAscii(ROW, 100);
-
- @Rule
- public TestName name = new TestName();
-
- @BeforeClass
- public static void setUpBeforeClass() throws Exception {
- conf1 = HBaseConfiguration.create();
- conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
- conf1.setLong("replication.source.sleepforretries", 100);
- // Each WAL is about 120 bytes
- conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, 200);
- conf1.setLong("replication.source.per.peer.node.bandwidth", 100L);
-
- utility1 = new HBaseTestingUtility(conf1);
- utility1.startMiniZKCluster();
- MiniZooKeeperCluster miniZK = utility1.getZkCluster();
- new ZKWatcher(conf1, "cluster1", null, true);
-
- conf2 = new Configuration(conf1);
- conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
-
- utility2 = new HBaseTestingUtility(conf2);
- utility2.setZkCluster(miniZK);
- new ZKWatcher(conf2, "cluster2", null, true);
-
- ReplicationAdmin admin1 = new ReplicationAdmin(conf1);
- ReplicationPeerConfig rpc = new ReplicationPeerConfig();
- rpc.setClusterKey(utility2.getClusterKey());
-
- utility1.startMiniCluster(1, 1);
- utility2.startMiniCluster(1, 1);
-
- admin1.addPeer("peer1", rpc, null);
- admin1.addPeer("peer2", rpc, null);
- admin1.addPeer("peer3", rpc, null);
- }
-
- @AfterClass
- public static void tearDownAfterClass() throws Exception {
- utility2.shutdownMiniCluster();
- utility1.shutdownMiniCluster();
- }
-
-
- volatile private boolean testQuotaPass = false;
- volatile private boolean testQuotaNonZero = false;
- @Test
- public void testQuota() throws IOException {
- final TableName tableName = TableName.valueOf(name.getMethodName());
- HTableDescriptor table = new HTableDescriptor(tableName);
- HColumnDescriptor fam = new HColumnDescriptor(famName);
- fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
- table.addFamily(fam);
- utility1.getAdmin().createTable(table);
- utility2.getAdmin().createTable(table);
-
- Thread watcher = new Thread(()->{
- Replication replication = (Replication)utility1.getMiniHBaseCluster()
- .getRegionServer(0).getReplicationSourceService();
- AtomicLong bufferUsed = replication.getReplicationManager().getTotalBufferUsed();
- testQuotaPass = true;
- while (!Thread.interrupted()) {
- long size = bufferUsed.get();
- if (size > 0) {
- testQuotaNonZero = true;
- }
- if (size > 600) {
- // We read logs first then check throttler, so if the buffer quota limiter doesn't
- // take effect, it will push many logs and exceed the quota.
- testQuotaPass = false;
- }
- Threads.sleep(50);
- }
- });
- watcher.start();
-
- try(Table t1 = utility1.getConnection().getTable(tableName);
- Table t2 = utility2.getConnection().getTable(tableName)) {
- for (int i = 0; i < 50; i++) {
- Put put = new Put(ROWS[i]);
- put.addColumn(famName, VALUE, VALUE);
- t1.put(put);
- }
- long start = EnvironmentEdgeManager.currentTime();
- while (EnvironmentEdgeManager.currentTime() - start < 180000) {
- Scan scan = new Scan();
- scan.setCaching(50);
- int count = 0;
- try (ResultScanner results = t2.getScanner(scan)) {
- for (Result result : results) {
- count++;
- }
- }
- if (count < 50) {
- LOG.info("Waiting all logs pushed to slave. Expected 50 , actual " + count);
- Threads.sleep(200);
- continue;
- }
- break;
- }
- }
-
- watcher.interrupt();
- Assert.assertTrue(testQuotaPass);
- Assert.assertTrue(testQuotaNonZero);
- }
-
- private List<Integer> getRowNumbers(List<Cell> cells) {
- List<Integer> listOfRowNumbers = new ArrayList<>(cells.size());
- for (Cell c : cells) {
- listOfRowNumbers.add(Integer.parseInt(Bytes
- .toString(c.getRowArray(), c.getRowOffset() + ROW.length,
- c.getRowLength() - ROW.length)));
- }
- return listOfRowNumbers;
- }
-}
[2/2] hbase git commit: HBASE-20667 Rename TestGlobalThrottler to
TestReplicationGlobalThrottler
Posted by ap...@apache.org.
HBASE-20667 Rename TestGlobalThrottler to TestReplicationGlobalThrottler
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/07487cb0
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/07487cb0
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/07487cb0
Branch: refs/heads/branch-1
Commit: 07487cb0b16a979bb56457701812bf38143fe16e
Parents: a25878c
Author: Andrew Purtell <ap...@apache.org>
Authored: Fri Jun 1 16:44:11 2018 -0700
Committer: Andrew Purtell <ap...@apache.org>
Committed: Fri Jun 1 16:47:26 2018 -0700
----------------------------------------------------------------------
.../TestGlobalReplicationThrottler.java | 175 +++++++++++++++++++
.../regionserver/TestGlobalThrottler.java | 175 -------------------
2 files changed, 175 insertions(+), 175 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/07487cb0/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java
new file mode 100644
index 0000000..641ea50
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalReplicationThrottler.java
@@ -0,0 +1,175 @@
+/*
+ *
+ * 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.hadoop.hbase.replication.regionserver;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.HTestConst;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+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.client.replication.ReplicationAdmin;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category({ ReplicationTests.class, LargeTests.class })
+public class TestGlobalReplicationThrottler {
+ private static final Log LOG = LogFactory.getLog(TestGlobalReplicationThrottler.class);
+ private static final int REPLICATION_SOURCE_QUOTA = 200;
+ private static int numOfPeer = 0;
+ private static Configuration conf1;
+ private static Configuration conf2;
+
+ private static HBaseTestingUtility utility1;
+ private static HBaseTestingUtility utility2;
+
+ private static final byte[] famName = Bytes.toBytes("f");
+ private static final byte[] VALUE = Bytes.toBytes("v");
+ private static final byte[] ROW = Bytes.toBytes("r");
+ private static final byte[][] ROWS = HTestConst.makeNAscii(ROW, 100);
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ conf1 = HBaseConfiguration.create();
+ conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
+ conf1.setLong("replication.source.sleepforretries", 100);
+ // Each WAL is about 120 bytes
+ conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, 200);
+ conf1.setLong("replication.source.per.peer.node.bandwidth", 100L);
+
+ utility1 = new HBaseTestingUtility(conf1);
+ utility1.startMiniZKCluster();
+ MiniZooKeeperCluster miniZK = utility1.getZkCluster();
+ new ZooKeeperWatcher(conf1, "cluster1", null, true);
+
+ conf2 = new Configuration(conf1);
+ conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
+
+ utility2 = new HBaseTestingUtility(conf2);
+ utility2.setZkCluster(miniZK);
+ new ZooKeeperWatcher(conf2, "cluster2", null, true);
+
+ ReplicationAdmin admin1 = new ReplicationAdmin(conf1);
+ ReplicationPeerConfig rpc = new ReplicationPeerConfig();
+ rpc.setClusterKey(utility2.getClusterKey());
+ admin1.addPeer("peer1", rpc, null);
+ admin1.addPeer("peer2", rpc, null);
+ admin1.addPeer("peer3", rpc, null);
+ numOfPeer = admin1.getPeersCount();
+
+ utility1.startMiniCluster(1, 1);
+ utility2.startMiniCluster(1, 1);
+ }
+
+ @AfterClass
+ public static void tearDownAfterClass() throws Exception {
+ utility2.shutdownMiniCluster();
+ utility1.shutdownMiniCluster();
+ }
+
+ volatile private boolean testQuotaPass = false;
+
+ @Test
+ public void testQuota() throws IOException {
+ TableName tableName = TableName.valueOf("testQuota");
+ HTableDescriptor table = new HTableDescriptor(tableName);
+ HColumnDescriptor fam = new HColumnDescriptor(famName);
+ fam.setScope(HConstants.REPLICATION_SCOPE_LOCAL);
+ table.addFamily(fam);
+ utility1.getHBaseAdmin().createTable(table);
+ utility2.getHBaseAdmin().createTable(table);
+
+ Thread watcher = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ Replication replication = (Replication) utility1.getMiniHBaseCluster()
+ .getRegionServer(0).getReplicationSourceService();
+ AtomicLong bufferUsed = replication.getReplicationManager().getTotalBufferUsed();
+ testQuotaPass = true;
+ while (!Thread.interrupted()) {
+ long size = bufferUsed.get();
+ //the reason here doing "numOfPeer + 1" is because by using method addEntryToBatch(), even the
+ // batch size (after added last entry) exceeds quota, it still keeps the last one in the batch
+ // so total used buffer size can be one "replication.total.buffer.quota" larger than expected
+ if (size > REPLICATION_SOURCE_QUOTA * (numOfPeer + 1)) {
+ // We read logs first then check throttler, so if the buffer quota limiter doesn't
+ // take effect, it will push many logs and exceed the quota.
+ testQuotaPass = false;
+ }
+ Threads.sleep(50);
+ }
+ }
+ });
+
+ watcher.start();
+
+ try(Table t1 = utility1.getConnection().getTable(tableName);
+ Table t2 = utility2.getConnection().getTable(tableName)) {
+ for (int i = 0; i < 50; i++) {
+ Put put = new Put(ROWS[i]);
+ put.addColumn(famName, VALUE, VALUE);
+ t1.put(put);
+ }
+ long start = EnvironmentEdgeManager.currentTime();
+ while (EnvironmentEdgeManager.currentTime() - start < 180000) {
+ Scan scan = new Scan();
+ scan.setCaching(50);
+ int count = 0;
+ try (ResultScanner results = t2.getScanner(scan)) {
+ for (Result result : results) {
+ count++;
+ }
+ }
+ if (count < 50) {
+ LOG.info("Waiting all logs pushed to slave. Expected 50 , actual " + count);
+ Threads.sleep(200);
+ continue;
+ }
+ break;
+ }
+ }
+
+ watcher.interrupt();
+ Assert.assertTrue(testQuotaPass);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/07487cb0/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
deleted file mode 100644
index 6e8f1cd..0000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
+++ /dev/null
@@ -1,175 +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.hadoop.hbase.replication.regionserver;
-
-import java.io.IOException;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HColumnDescriptor;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.HTestConst;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
-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.client.replication.ReplicationAdmin;
-import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
-import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.apache.hadoop.hbase.testclassification.ReplicationTests;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.hadoop.hbase.util.Threads;
-import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
-import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-@Category({ ReplicationTests.class, LargeTests.class })
-public class TestGlobalThrottler {
- private static final Log LOG = LogFactory.getLog(TestGlobalThrottler.class);
- private static final int REPLICATION_SOURCE_QUOTA = 200;
- private static int numOfPeer = 0;
- private static Configuration conf1;
- private static Configuration conf2;
-
- private static HBaseTestingUtility utility1;
- private static HBaseTestingUtility utility2;
-
- private static final byte[] famName = Bytes.toBytes("f");
- private static final byte[] VALUE = Bytes.toBytes("v");
- private static final byte[] ROW = Bytes.toBytes("r");
- private static final byte[][] ROWS = HTestConst.makeNAscii(ROW, 100);
-
- @BeforeClass
- public static void setUpBeforeClass() throws Exception {
- conf1 = HBaseConfiguration.create();
- conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
- conf1.setLong("replication.source.sleepforretries", 100);
- // Each WAL is about 120 bytes
- conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, 200);
- conf1.setLong("replication.source.per.peer.node.bandwidth", 100L);
-
- utility1 = new HBaseTestingUtility(conf1);
- utility1.startMiniZKCluster();
- MiniZooKeeperCluster miniZK = utility1.getZkCluster();
- new ZooKeeperWatcher(conf1, "cluster1", null, true);
-
- conf2 = new Configuration(conf1);
- conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2");
-
- utility2 = new HBaseTestingUtility(conf2);
- utility2.setZkCluster(miniZK);
- new ZooKeeperWatcher(conf2, "cluster2", null, true);
-
- ReplicationAdmin admin1 = new ReplicationAdmin(conf1);
- ReplicationPeerConfig rpc = new ReplicationPeerConfig();
- rpc.setClusterKey(utility2.getClusterKey());
- admin1.addPeer("peer1", rpc, null);
- admin1.addPeer("peer2", rpc, null);
- admin1.addPeer("peer3", rpc, null);
- numOfPeer = admin1.getPeersCount();
-
- utility1.startMiniCluster(1, 1);
- utility2.startMiniCluster(1, 1);
- }
-
- @AfterClass
- public static void tearDownAfterClass() throws Exception {
- utility2.shutdownMiniCluster();
- utility1.shutdownMiniCluster();
- }
-
- volatile private boolean testQuotaPass = false;
-
- @Test
- public void testQuota() throws IOException {
- TableName tableName = TableName.valueOf("testQuota");
- HTableDescriptor table = new HTableDescriptor(tableName);
- HColumnDescriptor fam = new HColumnDescriptor(famName);
- fam.setScope(HConstants.REPLICATION_SCOPE_LOCAL);
- table.addFamily(fam);
- utility1.getHBaseAdmin().createTable(table);
- utility2.getHBaseAdmin().createTable(table);
-
- Thread watcher = new Thread(new Runnable() {
- @Override
- public void run() {
- Replication replication = (Replication) utility1.getMiniHBaseCluster()
- .getRegionServer(0).getReplicationSourceService();
- AtomicLong bufferUsed = replication.getReplicationManager().getTotalBufferUsed();
- testQuotaPass = true;
- while (!Thread.interrupted()) {
- long size = bufferUsed.get();
- //the reason here doing "numOfPeer + 1" is because by using method addEntryToBatch(), even the
- // batch size (after added last entry) exceeds quota, it still keeps the last one in the batch
- // so total used buffer size can be one "replication.total.buffer.quota" larger than expected
- if (size > REPLICATION_SOURCE_QUOTA * (numOfPeer + 1)) {
- // We read logs first then check throttler, so if the buffer quota limiter doesn't
- // take effect, it will push many logs and exceed the quota.
- testQuotaPass = false;
- }
- Threads.sleep(50);
- }
- }
- });
-
- watcher.start();
-
- try(Table t1 = utility1.getConnection().getTable(tableName);
- Table t2 = utility2.getConnection().getTable(tableName)) {
- for (int i = 0; i < 50; i++) {
- Put put = new Put(ROWS[i]);
- put.addColumn(famName, VALUE, VALUE);
- t1.put(put);
- }
- long start = EnvironmentEdgeManager.currentTime();
- while (EnvironmentEdgeManager.currentTime() - start < 180000) {
- Scan scan = new Scan();
- scan.setCaching(50);
- int count = 0;
- try (ResultScanner results = t2.getScanner(scan)) {
- for (Result result : results) {
- count++;
- }
- }
- if (count < 50) {
- LOG.info("Waiting all logs pushed to slave. Expected 50 , actual " + count);
- Threads.sleep(200);
- continue;
- }
- break;
- }
- }
-
- watcher.interrupt();
- Assert.assertTrue(testQuotaPass);
- }
-}