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.EMPTY_START_ROW; 021import static org.apache.hadoop.hbase.HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT; 022import static org.apache.hadoop.hbase.coprocessor.CoprocessorHost.REGION_COPROCESSOR_CONF_KEY; 023import static org.hamcrest.CoreMatchers.instanceOf; 024import static org.junit.Assert.assertEquals; 025import static org.junit.Assert.assertThat; 026import static org.junit.Assert.assertTrue; 027import static org.junit.Assert.fail; 028 029import java.io.IOException; 030import java.util.Optional; 031import java.util.concurrent.CompletionException; 032import java.util.concurrent.ExecutionException; 033import java.util.concurrent.TimeUnit; 034import java.util.concurrent.atomic.AtomicReference; 035import org.apache.commons.io.IOUtils; 036import org.apache.hadoop.conf.Configuration; 037import org.apache.hadoop.hbase.HBaseClassTestRule; 038import org.apache.hadoop.hbase.HBaseTestingUtility; 039import org.apache.hadoop.hbase.HRegionLocation; 040import org.apache.hadoop.hbase.TableName; 041import org.apache.hadoop.hbase.TableNotFoundException; 042import org.apache.hadoop.hbase.coprocessor.ObserverContext; 043import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor; 044import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; 045import org.apache.hadoop.hbase.coprocessor.RegionObserver; 046import org.apache.hadoop.hbase.exceptions.TimeoutIOException; 047import org.apache.hadoop.hbase.security.User; 048import org.apache.hadoop.hbase.testclassification.ClientTests; 049import org.apache.hadoop.hbase.testclassification.MediumTests; 050import org.apache.hadoop.hbase.util.Bytes; 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; 057 058@Category({ MediumTests.class, ClientTests.class }) 059public class TestAsyncRegionLocator { 060 061 @ClassRule 062 public static final HBaseClassTestRule CLASS_RULE = 063 HBaseClassTestRule.forClass(TestAsyncRegionLocator.class); 064 065 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); 066 067 private static TableName TABLE_NAME = TableName.valueOf("async"); 068 069 private static byte[] FAMILY = Bytes.toBytes("cf"); 070 071 private static AsyncConnectionImpl CONN; 072 073 private static AsyncRegionLocator LOCATOR; 074 075 private static volatile long SLEEP_MS = 0L; 076 077 public static class SleepRegionObserver implements RegionCoprocessor, RegionObserver { 078 @Override 079 public Optional<RegionObserver> getRegionObserver() { 080 return Optional.of(this); 081 } 082 083 @Override 084 public void preScannerOpen(ObserverContext<RegionCoprocessorEnvironment> e, Scan scan) 085 throws IOException { 086 if (SLEEP_MS > 0) { 087 Threads.sleepWithoutInterrupt(SLEEP_MS); 088 } 089 } 090 } 091 092 @BeforeClass 093 public static void setUp() throws Exception { 094 Configuration conf = TEST_UTIL.getConfiguration(); 095 conf.set(REGION_COPROCESSOR_CONF_KEY, SleepRegionObserver.class.getName()); 096 conf.setLong(HBASE_CLIENT_META_OPERATION_TIMEOUT, 2000); 097 TEST_UTIL.startMiniCluster(1); 098 TEST_UTIL.createTable(TABLE_NAME, FAMILY); 099 TEST_UTIL.waitTableAvailable(TABLE_NAME); 100 AsyncRegistry registry = AsyncRegistryFactory.getRegistry(TEST_UTIL.getConfiguration()); 101 CONN = new AsyncConnectionImpl(TEST_UTIL.getConfiguration(), registry, 102 registry.getClusterId().get(), User.getCurrent()); 103 LOCATOR = CONN.getLocator(); 104 } 105 106 @AfterClass 107 public static void tearDown() throws Exception { 108 IOUtils.closeQuietly(CONN); 109 TEST_UTIL.shutdownMiniCluster(); 110 } 111 112 @Test 113 public void testTimeout() throws InterruptedException, ExecutionException { 114 SLEEP_MS = 1000; 115 long startNs = System.nanoTime(); 116 try { 117 LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, RegionLocateType.CURRENT, 118 TimeUnit.MILLISECONDS.toNanos(500)).get(); 119 fail(); 120 } catch (ExecutionException e) { 121 assertThat(e.getCause(), instanceOf(TimeoutIOException.class)); 122 } 123 long costMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs); 124 assertTrue(costMs >= 500); 125 assertTrue(costMs < 1000); 126 // wait for the background task finish 127 Thread.sleep(2000); 128 // Now the location should be in cache, so we will not visit meta again. 129 HRegionLocation loc = LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, 130 RegionLocateType.CURRENT, TimeUnit.MILLISECONDS.toNanos(500)).get(); 131 assertEquals(loc.getServerName(), 132 TEST_UTIL.getHBaseCluster().getRegionServer(0).getServerName()); 133 } 134 135 @Test 136 public void testNoCompletionException() { 137 // make sure that we do not get CompletionException 138 SLEEP_MS = 0; 139 AtomicReference<Throwable> errorHolder = new AtomicReference<>(); 140 try { 141 LOCATOR.getRegionLocation(TableName.valueOf("NotExist"), EMPTY_START_ROW, 142 RegionLocateType.CURRENT, TimeUnit.SECONDS.toNanos(1)) 143 .whenComplete((r, e) -> errorHolder.set(e)).join(); 144 fail(); 145 } catch (CompletionException e) { 146 // join will return a CompletionException, which is OK 147 assertThat(e.getCause(), instanceOf(TableNotFoundException.class)); 148 } 149 // but we need to make sure that we do not get a CompletionException in the callback 150 assertThat(errorHolder.get(), instanceOf(TableNotFoundException.class)); 151 } 152}