001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.replication.regionserver; 019 020import java.io.IOException; 021import org.apache.hadoop.conf.Configuration; 022import org.apache.hadoop.hbase.HBaseClassTestRule; 023import org.apache.hadoop.hbase.HBaseConfiguration; 024import org.apache.hadoop.hbase.HBaseTestingUtil; 025import org.apache.hadoop.hbase.HConstants; 026import org.apache.hadoop.hbase.HTestConst; 027import org.apache.hadoop.hbase.TableName; 028import org.apache.hadoop.hbase.client.Admin; 029import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 030import org.apache.hadoop.hbase.client.Connection; 031import org.apache.hadoop.hbase.client.ConnectionFactory; 032import org.apache.hadoop.hbase.client.Put; 033import org.apache.hadoop.hbase.client.Result; 034import org.apache.hadoop.hbase.client.ResultScanner; 035import org.apache.hadoop.hbase.client.Scan; 036import org.apache.hadoop.hbase.client.Table; 037import org.apache.hadoop.hbase.client.TableDescriptor; 038import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 039import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; 040import org.apache.hadoop.hbase.testclassification.LargeTests; 041import org.apache.hadoop.hbase.testclassification.ReplicationTests; 042import org.apache.hadoop.hbase.util.Bytes; 043import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 044import org.apache.hadoop.hbase.util.Threads; 045import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; 046import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 047import org.junit.AfterClass; 048import org.junit.Assert; 049import org.junit.BeforeClass; 050import org.junit.ClassRule; 051import org.junit.Rule; 052import org.junit.Test; 053import org.junit.experimental.categories.Category; 054import org.junit.rules.TestName; 055import org.slf4j.Logger; 056import org.slf4j.LoggerFactory; 057 058@Category({ ReplicationTests.class, LargeTests.class }) 059public class TestGlobalReplicationThrottler { 060 061 @ClassRule 062 public static final HBaseClassTestRule CLASS_RULE = 063 HBaseClassTestRule.forClass(TestGlobalReplicationThrottler.class); 064 065 private static final Logger LOG = LoggerFactory.getLogger(TestGlobalReplicationThrottler.class); 066 private static final int REPLICATION_SOURCE_QUOTA = 200; 067 private static int numOfPeer = 0; 068 private static Configuration conf1; 069 private static Configuration conf2; 070 071 private static HBaseTestingUtil utility1; 072 private static HBaseTestingUtil utility2; 073 074 private static final byte[] famName = Bytes.toBytes("f"); 075 private static final byte[] VALUE = Bytes.toBytes("v"); 076 private static final byte[] ROW = Bytes.toBytes("r"); 077 private static final byte[][] ROWS = HTestConst.makeNAscii(ROW, 100); 078 079 @Rule 080 public TestName name = new TestName(); 081 082 @BeforeClass 083 public static void setUpBeforeClass() throws Exception { 084 conf1 = HBaseConfiguration.create(); 085 conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1"); 086 conf1.setLong("replication.source.sleepforretries", 100); 087 // Each WAL is about 120 bytes 088 conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, REPLICATION_SOURCE_QUOTA); 089 conf1.setLong("replication.source.per.peer.node.bandwidth", 100L); 090 091 utility1 = new HBaseTestingUtil(conf1); 092 utility1.startMiniZKCluster(); 093 MiniZooKeeperCluster miniZK = utility1.getZkCluster(); 094 new ZKWatcher(conf1, "cluster1", null, true); 095 096 conf2 = new Configuration(conf1); 097 conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2"); 098 099 utility2 = new HBaseTestingUtil(conf2); 100 utility2.setZkCluster(miniZK); 101 new ZKWatcher(conf2, "cluster2", null, true); 102 103 ReplicationPeerConfig rpc = 104 ReplicationPeerConfig.newBuilder().setClusterKey(utility2.getClusterKey()).build(); 105 106 utility1.startMiniCluster(); 107 utility2.startMiniCluster(); 108 109 try (Connection connection = ConnectionFactory.createConnection(utility1.getConfiguration()); 110 Admin admin1 = connection.getAdmin()) { 111 admin1.addReplicationPeer("peer1", rpc); 112 admin1.addReplicationPeer("peer2", rpc); 113 admin1.addReplicationPeer("peer3", rpc); 114 numOfPeer = admin1.listReplicationPeers().size(); 115 } 116 } 117 118 @AfterClass 119 public static void tearDownAfterClass() throws Exception { 120 utility2.shutdownMiniCluster(); 121 utility1.shutdownMiniCluster(); 122 } 123 124 volatile private boolean testQuotaPass = false; 125 volatile private boolean testQuotaNonZero = false; 126 127 @Test 128 public void testQuota() throws IOException { 129 final TableName tableName = TableName.valueOf(name.getMethodName()); 130 TableDescriptor tableDescriptor = 131 TableDescriptorBuilder.newBuilder(tableName).setColumnFamily(ColumnFamilyDescriptorBuilder 132 .newBuilder(famName).setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build()).build(); 133 utility1.getAdmin().createTable(tableDescriptor); 134 utility2.getAdmin().createTable(tableDescriptor); 135 136 Thread watcher = new Thread(() -> { 137 Replication replication = (Replication) utility1.getMiniHBaseCluster().getRegionServer(0) 138 .getReplicationSourceService(); 139 testQuotaPass = true; 140 while (!Thread.interrupted()) { 141 long size = replication.getReplicationManager().getTotalBufferUsed(); 142 if (size > 0) { 143 testQuotaNonZero = true; 144 } 145 // the reason here doing "numOfPeer + 1" is because by using method addEntryToBatch(), even 146 // the 147 // batch size (after added last entry) exceeds quota, it still keeps the last one in the 148 // batch 149 // so total used buffer size can be one "replication.total.buffer.quota" larger than 150 // expected 151 if (size > REPLICATION_SOURCE_QUOTA * (numOfPeer + 1)) { 152 // We read logs first then check throttler, so if the buffer quota limiter doesn't 153 // take effect, it will push many logs and exceed the quota. 154 testQuotaPass = false; 155 } 156 Threads.sleep(50); 157 } 158 }); 159 watcher.start(); 160 161 try (Table t1 = utility1.getConnection().getTable(tableName); 162 Table t2 = utility2.getConnection().getTable(tableName)) { 163 for (int i = 0; i < 50; i++) { 164 Put put = new Put(ROWS[i]); 165 put.addColumn(famName, VALUE, VALUE); 166 t1.put(put); 167 } 168 long start = EnvironmentEdgeManager.currentTime(); 169 while (EnvironmentEdgeManager.currentTime() - start < 180000) { 170 Scan scan = new Scan(); 171 scan.setCaching(50); 172 int count = 0; 173 try (ResultScanner results = t2.getScanner(scan)) { 174 for (Result result : results) { 175 count++; 176 } 177 } 178 if (count < 50) { 179 LOG.info("Waiting all logs pushed to slave. Expected 50 , actual " + count); 180 Threads.sleep(200); 181 continue; 182 } 183 break; 184 } 185 } 186 187 watcher.interrupt(); 188 Assert.assertTrue(testQuotaPass); 189 Assert.assertTrue(testQuotaNonZero); 190 } 191 192}