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.client; 019 020import static org.apache.hadoop.hbase.HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT; 021import static org.apache.hadoop.hbase.io.ByteBuffAllocator.MAX_BUFFER_COUNT_KEY; 022import static org.apache.hadoop.hbase.master.LoadBalancer.TABLES_ON_MASTER; 023import static org.junit.Assert.assertEquals; 024 025import java.io.IOException; 026import java.util.ArrayList; 027import java.util.Arrays; 028import java.util.Collections; 029import java.util.List; 030import java.util.Random; 031import java.util.concurrent.ExecutionException; 032import java.util.concurrent.ExecutorService; 033import java.util.concurrent.Executors; 034import java.util.concurrent.Future; 035import java.util.concurrent.TimeUnit; 036import java.util.concurrent.atomic.AtomicBoolean; 037import java.util.stream.Collectors; 038import java.util.stream.IntStream; 039import org.apache.hadoop.hbase.HBaseClassTestRule; 040import org.apache.hadoop.hbase.HBaseTestingUtility; 041import org.apache.hadoop.hbase.MemoryCompactionPolicy; 042import org.apache.hadoop.hbase.ServerName; 043import org.apache.hadoop.hbase.TableName; 044import org.apache.hadoop.hbase.Waiter.ExplainingPredicate; 045import org.apache.hadoop.hbase.regionserver.CompactingMemStore; 046import org.apache.hadoop.hbase.regionserver.HRegion; 047import org.apache.hadoop.hbase.testclassification.ClientTests; 048import org.apache.hadoop.hbase.testclassification.LargeTests; 049import org.apache.hadoop.hbase.util.Bytes; 050import org.apache.hadoop.hbase.util.RetryCounter; 051import org.apache.hadoop.hbase.util.Threads; 052import org.junit.AfterClass; 053import org.junit.BeforeClass; 054import org.junit.ClassRule; 055import org.junit.Test; 056import org.junit.experimental.categories.Category; 057import org.slf4j.Logger; 058import org.slf4j.LoggerFactory; 059 060import org.apache.hbase.thirdparty.com.google.common.io.Closeables; 061 062/** 063 * Will split the table, and move region randomly when testing. 064 */ 065@Category({ LargeTests.class, ClientTests.class }) 066public class TestAsyncTableGetMultiThreaded { 067 068 @ClassRule 069 public static final HBaseClassTestRule CLASS_RULE = 070 HBaseClassTestRule.forClass(TestAsyncTableGetMultiThreaded.class); 071 072 private static final Logger LOG = LoggerFactory.getLogger(TestAsyncTableGetMultiThreaded.class); 073 074 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); 075 076 private static TableName TABLE_NAME = TableName.valueOf("async"); 077 078 private static byte[] FAMILY = Bytes.toBytes("cf"); 079 080 private static byte[] QUALIFIER = Bytes.toBytes("cq"); 081 082 private static int COUNT = 1000; 083 084 private static AsyncConnection CONN; 085 086 private static AsyncTable<?> TABLE; 087 088 private static byte[][] SPLIT_KEYS; 089 090 @BeforeClass 091 public static void setUp() throws Exception { 092 setUp(MemoryCompactionPolicy.NONE); 093 } 094 095 protected static void setUp(MemoryCompactionPolicy memoryCompaction) throws Exception { 096 TEST_UTIL.getConfiguration().set(TABLES_ON_MASTER, "none"); 097 TEST_UTIL.getConfiguration().setLong(HBASE_CLIENT_META_OPERATION_TIMEOUT, 60000L); 098 TEST_UTIL.getConfiguration().setInt(MAX_BUFFER_COUNT_KEY, 100); 099 TEST_UTIL.getConfiguration().set(CompactingMemStore.COMPACTING_MEMSTORE_TYPE_KEY, 100 String.valueOf(memoryCompaction)); 101 102 TEST_UTIL.startMiniCluster(3); 103 SPLIT_KEYS = new byte[8][]; 104 for (int i = 111; i < 999; i += 111) { 105 SPLIT_KEYS[i / 111 - 1] = Bytes.toBytes(String.format("%03d", i)); 106 } 107 TEST_UTIL.createTable(TABLE_NAME, FAMILY); 108 TEST_UTIL.waitTableAvailable(TABLE_NAME); 109 CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get(); 110 TABLE = CONN.getTableBuilder(TABLE_NAME).setReadRpcTimeout(1, TimeUnit.SECONDS) 111 .setMaxRetries(1000).build(); 112 TABLE.putAll( 113 IntStream.range(0, COUNT).mapToObj(i -> new Put(Bytes.toBytes(String.format("%03d", i))) 114 .addColumn(FAMILY, QUALIFIER, Bytes.toBytes(i))).collect(Collectors.toList())) 115 .get(); 116 } 117 118 @AfterClass 119 public static void tearDown() throws Exception { 120 Closeables.close(CONN, true); 121 TEST_UTIL.shutdownMiniCluster(); 122 } 123 124 private void run(AtomicBoolean stop) throws InterruptedException, ExecutionException { 125 while (!stop.get()) { 126 for (int i = 0; i < COUNT; i++) { 127 assertEquals(i, Bytes.toInt(TABLE.get(new Get(Bytes.toBytes(String.format("%03d", i)))) 128 .get().getValue(FAMILY, QUALIFIER))); 129 } 130 // sleep a bit so we do not add to much load to the test machine as we have 20 threads here 131 Thread.sleep(10); 132 } 133 } 134 135 @Test 136 public void test() throws Exception { 137 LOG.info("====== Test started ======"); 138 int numThreads = 7; 139 AtomicBoolean stop = new AtomicBoolean(false); 140 ExecutorService executor = 141 Executors.newFixedThreadPool(numThreads, Threads.newDaemonThreadFactory("TestAsyncGet-")); 142 List<Future<?>> futures = new ArrayList<>(); 143 IntStream.range(0, numThreads).forEach(i -> futures.add(executor.submit(() -> { 144 run(stop); 145 return null; 146 }))); 147 LOG.info("====== Scheduled {} read threads ======", numThreads); 148 Collections.shuffle(Arrays.asList(SPLIT_KEYS), new Random(123)); 149 Admin admin = TEST_UTIL.getAdmin(); 150 for (byte[] splitPoint : SPLIT_KEYS) { 151 int oldRegionCount = admin.getRegions(TABLE_NAME).size(); 152 LOG.info("====== Splitting at {} ======, region count before splitting is {}", 153 Bytes.toStringBinary(splitPoint), oldRegionCount); 154 admin.split(TABLE_NAME, splitPoint); 155 TEST_UTIL.waitFor(30000, new ExplainingPredicate<Exception>() { 156 @Override 157 public boolean evaluate() throws Exception { 158 return TEST_UTIL.getMiniHBaseCluster().getRegions(TABLE_NAME).size() > oldRegionCount; 159 } 160 161 @Override 162 public String explainFailure() throws Exception { 163 return "Split has not finished yet"; 164 } 165 }); 166 List<HRegion> regions = TEST_UTIL.getMiniHBaseCluster().getRegions(TABLE_NAME); 167 LOG.info("====== Split at {} ======, region count after splitting is {}", 168 Bytes.toStringBinary(splitPoint), regions.size()); 169 for (HRegion region : regions) { 170 LOG.info("====== Compact {} ======", region.getRegionInfo()); 171 region.compact(true); 172 } 173 for (HRegion region : regions) { 174 // Waiting for compaction to complete and references are cleaned up 175 LOG.info("====== Waiting for compaction on {} ======", region.getRegionInfo()); 176 RetryCounter retrier = new RetryCounter(30, 1, TimeUnit.SECONDS); 177 for (;;) { 178 try { 179 if (admin.getCompactionStateForRegion( 180 region.getRegionInfo().getRegionName()) == CompactionState.NONE) { 181 break; 182 } 183 } catch (IOException e) { 184 LOG.warn("Failed to query"); 185 } 186 if (!retrier.shouldRetry()) { 187 throw new IOException("Can not finish compaction in time after attempt " + 188 retrier.getAttemptTimes() + " times"); 189 } 190 retrier.sleepUntilNextRetry(); 191 } 192 LOG.info("====== Compaction on {} finished, close and archive compacted files ======", 193 region.getRegionInfo()); 194 region.getStores().get(0).closeAndArchiveCompactedFiles(); 195 LOG.info("====== Close and archive compacted files on {} done ======", 196 region.getRegionInfo()); 197 } 198 Thread.sleep(5000); 199 LOG.info("====== Balancing cluster ======"); 200 admin.balance(true); 201 LOG.info("====== Balance cluster done ======"); 202 Thread.sleep(5000); 203 ServerName metaServer = TEST_UTIL.getHBaseCluster().getServerHoldingMeta(); 204 ServerName newMetaServer = TEST_UTIL.getHBaseCluster().getRegionServerThreads().stream() 205 .map(t -> t.getRegionServer().getServerName()).filter(s -> !s.equals(metaServer)) 206 .findAny().get(); 207 LOG.info("====== Moving meta from {} to {} ======", metaServer, newMetaServer); 208 admin.move(RegionInfoBuilder.FIRST_META_REGIONINFO.getEncodedNameAsBytes(), newMetaServer); 209 LOG.info("====== Move meta done ======"); 210 Thread.sleep(5000); 211 } 212 LOG.info("====== Read test finished, shutdown thread pool ======"); 213 stop.set(true); 214 executor.shutdown(); 215 for (int i = 0; i < numThreads; i++) { 216 LOG.info("====== Waiting for {} threads to finish, remaining {} ======", numThreads, 217 numThreads - i); 218 futures.get(i).get(); 219 } 220 LOG.info("====== Test test finished ======"); 221 } 222}