Source code

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 java.util.stream.Collectors.toList;
021import static org.apache.hadoop.hbase.HConstants.EMPTY_END_ROW;
022import static org.apache.hadoop.hbase.HConstants.EMPTY_START_ROW;
023import static org.hamcrest.CoreMatchers.instanceOf;
024import static org.junit.Assert.assertArrayEquals;
025import static org.junit.Assert.assertEquals;
026import static org.junit.Assert.assertSame;
027import static org.junit.Assert.assertThat;
028
029import java.io.IOException;
030import java.util.Arrays;
031import java.util.List;
032import java.util.concurrent.CompletableFuture;
033import java.util.concurrent.ExecutionException;
034import java.util.concurrent.ThreadLocalRandom;
035import java.util.stream.IntStream;
036import org.apache.commons.io.IOUtils;
037import org.apache.hadoop.hbase.HBaseClassTestRule;
038import org.apache.hadoop.hbase.HBaseTestingUtility;
039import org.apache.hadoop.hbase.HRegionLocation;
040import org.apache.hadoop.hbase.NotServingRegionException;
041import org.apache.hadoop.hbase.ServerName;
042import org.apache.hadoop.hbase.TableName;
043import org.apache.hadoop.hbase.TableNotFoundException;
044import org.apache.hadoop.hbase.Waiter.ExplainingPredicate;
045import org.apache.hadoop.hbase.security.User;
046import org.apache.hadoop.hbase.testclassification.ClientTests;
047import org.apache.hadoop.hbase.testclassification.MediumTests;
048import org.apache.hadoop.hbase.util.Bytes;
049import org.junit.After;
050import org.junit.AfterClass;
051import org.junit.BeforeClass;
052import org.junit.ClassRule;
053import org.junit.Test;
054import org.junit.experimental.categories.Category;
055
056@Category({ MediumTests.class, ClientTests.class })
057public class TestAsyncNonMetaRegionLocator {
058
059  @ClassRule
060  public static final HBaseClassTestRule CLASS_RULE =
061      HBaseClassTestRule.forClass(TestAsyncNonMetaRegionLocator.class);
062
063  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
064
065  private static TableName TABLE_NAME = TableName.valueOf("async");
066
067  private static byte[] FAMILY = Bytes.toBytes("cf");
068
069  private static AsyncConnectionImpl CONN;
070
071  private static AsyncNonMetaRegionLocator LOCATOR;
072
073  private static byte[][] SPLIT_KEYS;
074
075  @BeforeClass
076  public static void setUp() throws Exception {
077    TEST_UTIL.startMiniCluster(3);
078    TEST_UTIL.getAdmin().balancerSwitch(false, true);
079    AsyncRegistry registry = AsyncRegistryFactory.getRegistry(TEST_UTIL.getConfiguration());
080    CONN = new AsyncConnectionImpl(TEST_UTIL.getConfiguration(), registry,
081        registry.getClusterId().get(), User.getCurrent());
082    LOCATOR = new AsyncNonMetaRegionLocator(CONN);
083    SPLIT_KEYS = new byte[8][];
084    for (int i = 111; i < 999; i += 111) {
085      SPLIT_KEYS[i / 111 - 1] = Bytes.toBytes(String.format("%03d", i));
086    }
087  }
088
089  @AfterClass
090  public static void tearDown() throws Exception {
091    IOUtils.closeQuietly(CONN);
092    TEST_UTIL.shutdownMiniCluster();
093  }
094
095  @After
096  public void tearDownAfterTest() throws IOException {
097    Admin admin = TEST_UTIL.getAdmin();
098    if (admin.tableExists(TABLE_NAME)) {
099      if (admin.isTableEnabled(TABLE_NAME)) {
100        TEST_UTIL.getAdmin().disableTable(TABLE_NAME);
101      }
102      TEST_UTIL.getAdmin().deleteTable(TABLE_NAME);
103    }
104    LOCATOR.clearCache(TABLE_NAME);
105  }
106
107  private void createSingleRegionTable() throws IOException, InterruptedException {
108    TEST_UTIL.createTable(TABLE_NAME, FAMILY);
109    TEST_UTIL.waitTableAvailable(TABLE_NAME);
110  }
111
112  @Test
113  public void testNoTable() throws InterruptedException {
114    for (RegionLocateType locateType : RegionLocateType.values()) {
115      try {
116        LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, locateType, false).get();
117      } catch (ExecutionException e) {
118        assertThat(e.getCause(), instanceOf(TableNotFoundException.class));
119      }
120    }
121  }
122
123  @Test
124  public void testDisableTable() throws IOException, InterruptedException {
125    createSingleRegionTable();
126    TEST_UTIL.getAdmin().disableTable(TABLE_NAME);
127    for (RegionLocateType locateType : RegionLocateType.values()) {
128      try {
129        LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, locateType, false).get();
130      } catch (ExecutionException e) {
131        assertThat(e.getCause(), instanceOf(TableNotFoundException.class));
132      }
133    }
134  }
135
136  private void assertLocEquals(byte[] startKey, byte[] endKey, ServerName serverName,
137      HRegionLocation loc) {
138    RegionInfo info = loc.getRegion();
139    assertEquals(TABLE_NAME, info.getTable());
140    assertArrayEquals(startKey, info.getStartKey());
141    assertArrayEquals(endKey, info.getEndKey());
142    assertEquals(serverName, loc.getServerName());
143  }
144
145  @Test
146  public void testSingleRegionTable() throws IOException, InterruptedException, ExecutionException {
147    createSingleRegionTable();
148    ServerName serverName = TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME).getServerName();
149    for (RegionLocateType locateType : RegionLocateType.values()) {
150      assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, serverName,
151        LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, locateType, false).get());
152    }
153    byte[] randKey = new byte[ThreadLocalRandom.current().nextInt(128)];
154    ThreadLocalRandom.current().nextBytes(randKey);
155    for (RegionLocateType locateType : RegionLocateType.values()) {
156      assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, serverName,
157        LOCATOR.getRegionLocation(TABLE_NAME, randKey, locateType, false).get());
158    }
159  }
160
161  private void createMultiRegionTable() throws IOException, InterruptedException {
162    TEST_UTIL.createTable(TABLE_NAME, FAMILY, SPLIT_KEYS);
163    TEST_UTIL.waitTableAvailable(TABLE_NAME);
164  }
165
166  private static byte[][] getStartKeys() {
167    byte[][] startKeys = new byte[SPLIT_KEYS.length + 1][];
168    startKeys[0] = EMPTY_START_ROW;
169    System.arraycopy(SPLIT_KEYS, 0, startKeys, 1, SPLIT_KEYS.length);
170    return startKeys;
171  }
172
173  private static byte[][] getEndKeys() {
174    byte[][] endKeys = Arrays.copyOf(SPLIT_KEYS, SPLIT_KEYS.length + 1);
175    endKeys[endKeys.length - 1] = EMPTY_START_ROW;
176    return endKeys;
177  }
178
179  private ServerName[] getLocations(byte[][] startKeys) {
180    ServerName[] serverNames = new ServerName[startKeys.length];
181    TEST_UTIL.getHBaseCluster().getRegionServerThreads().stream().map(t -> t.getRegionServer())
182        .forEach(rs -> {
183          rs.getRegions(TABLE_NAME).forEach(r -> {
184            serverNames[Arrays.binarySearch(startKeys, r.getRegionInfo().getStartKey(),
185              Bytes::compareTo)] = rs.getServerName();
186          });
187        });
188    return serverNames;
189  }
190
191  @Test
192  public void testMultiRegionTable() throws IOException, InterruptedException {
193    createMultiRegionTable();
194    byte[][] startKeys = getStartKeys();
195    ServerName[] serverNames = getLocations(startKeys);
196    IntStream.range(0, 2).forEach(n -> IntStream.range(0, startKeys.length).forEach(i -> {
197      try {
198        assertLocEquals(startKeys[i], i == startKeys.length - 1 ? EMPTY_END_ROW : startKeys[i + 1],
199          serverNames[i], LOCATOR
200              .getRegionLocation(TABLE_NAME, startKeys[i], RegionLocateType.CURRENT, false).get());
201      } catch (InterruptedException | ExecutionException e) {
202        throw new RuntimeException(e);
203      }
204    }));
205
206    LOCATOR.clearCache(TABLE_NAME);
207    IntStream.range(0, 2).forEach(n -> IntStream.range(0, startKeys.length).forEach(i -> {
208      try {
209        assertLocEquals(startKeys[i], i == startKeys.length - 1 ? EMPTY_END_ROW : startKeys[i + 1],
210          serverNames[i],
211          LOCATOR.getRegionLocation(TABLE_NAME, startKeys[i], RegionLocateType.AFTER, false).get());
212      } catch (InterruptedException | ExecutionException e) {
213        throw new RuntimeException(e);
214      }
215    }));
216
217    LOCATOR.clearCache(TABLE_NAME);
218    byte[][] endKeys = getEndKeys();
219    IntStream.range(0, 2).forEach(
220      n -> IntStream.range(0, endKeys.length).map(i -> endKeys.length - 1 - i).forEach(i -> {
221        try {
222          assertLocEquals(i == 0 ? EMPTY_START_ROW : endKeys[i - 1], endKeys[i], serverNames[i],
223            LOCATOR.getRegionLocation(TABLE_NAME, endKeys[i], RegionLocateType.BEFORE, false)
224                .get());
225        } catch (InterruptedException | ExecutionException e) {
226          throw new RuntimeException(e);
227        }
228      }));
229  }
230
231  @Test
232  public void testRegionMove() throws IOException, InterruptedException, ExecutionException {
233    createSingleRegionTable();
234    ServerName serverName = TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME).getServerName();
235    HRegionLocation loc = LOCATOR
236        .getRegionLocation(TABLE_NAME, EMPTY_START_ROW, RegionLocateType.CURRENT, false).get();
237    assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, serverName, loc);
238    ServerName newServerName = TEST_UTIL.getHBaseCluster().getRegionServerThreads().stream()
239        .map(t -> t.getRegionServer().getServerName()).filter(sn -> !sn.equals(serverName))
240        .findAny().get();
241
242    TEST_UTIL.getAdmin().move(Bytes.toBytes(loc.getRegion().getEncodedName()),
243      Bytes.toBytes(newServerName.getServerName()));
244    while (!TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME).getServerName()
245        .equals(newServerName)) {
246      Thread.sleep(100);
247    }
248    // Should be same as it is in cache
249    assertSame(loc, LOCATOR
250        .getRegionLocation(TABLE_NAME, EMPTY_START_ROW, RegionLocateType.CURRENT, false).get());
251    LOCATOR.updateCachedLocation(loc, null);
252    // null error will not trigger a cache cleanup
253    assertSame(loc, LOCATOR
254        .getRegionLocation(TABLE_NAME, EMPTY_START_ROW, RegionLocateType.CURRENT, false).get());
255    LOCATOR.updateCachedLocation(loc, new NotServingRegionException());
256    assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, newServerName, LOCATOR
257        .getRegionLocation(TABLE_NAME, EMPTY_START_ROW, RegionLocateType.CURRENT, false).get());
258  }
259
260  // usually locate after will return the same result, so we add a test to make it return different
261  // result.
262  @Test
263  public void testLocateAfter() throws IOException, InterruptedException, ExecutionException {
264    byte[] row = Bytes.toBytes("1");
265    byte[] splitKey = Arrays.copyOf(row, 2);
266    TEST_UTIL.createTable(TABLE_NAME, FAMILY, new byte[][] { splitKey });
267    TEST_UTIL.waitTableAvailable(TABLE_NAME);
268    HRegionLocation currentLoc =
269        LOCATOR.getRegionLocation(TABLE_NAME, row, RegionLocateType.CURRENT, false).get();
270    ServerName currentServerName = TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME).getServerName();
271    assertLocEquals(EMPTY_START_ROW, splitKey, currentServerName, currentLoc);
272
273    HRegionLocation afterLoc =
274        LOCATOR.getRegionLocation(TABLE_NAME, row, RegionLocateType.AFTER, false).get();
275    ServerName afterServerName =
276        TEST_UTIL.getHBaseCluster().getRegionServerThreads().stream().map(t -> t.getRegionServer())
277            .filter(rs -> rs.getRegions(TABLE_NAME).stream()
278                .anyMatch(r -> Bytes.equals(splitKey, r.getRegionInfo().getStartKey())))
279            .findAny().get().getServerName();
280    assertLocEquals(splitKey, EMPTY_END_ROW, afterServerName, afterLoc);
281
282    assertSame(afterLoc,
283      LOCATOR.getRegionLocation(TABLE_NAME, row, RegionLocateType.AFTER, false).get());
284  }
285
286  // For HBASE-17402
287  @Test
288  public void testConcurrentLocate() throws IOException, InterruptedException, ExecutionException {
289    createMultiRegionTable();
290    byte[][] startKeys = getStartKeys();
291    byte[][] endKeys = getEndKeys();
292    ServerName[] serverNames = getLocations(startKeys);
293    for (int i = 0; i < 100; i++) {
294      LOCATOR.clearCache(TABLE_NAME);
295      List<CompletableFuture<HRegionLocation>> futures = IntStream.range(0, 1000)
296          .mapToObj(n -> String.format("%03d", n)).map(s -> Bytes.toBytes(s))
297          .map(r -> LOCATOR.getRegionLocation(TABLE_NAME, r, RegionLocateType.CURRENT, false))
298          .collect(toList());
299      for (int j = 0; j < 1000; j++) {
300        int index = Math.min(8, j / 111);
301        assertLocEquals(startKeys[index], endKeys[index], serverNames[index], futures.get(j).get());
302      }
303    }
304  }
305
306  @Test
307  public void testReload() throws Exception {
308    createSingleRegionTable();
309    ServerName serverName = TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME).getServerName();
310    for (RegionLocateType locateType : RegionLocateType.values()) {
311      assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, serverName,
312        LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, locateType, false).get());
313    }
314    ServerName newServerName = TEST_UTIL.getHBaseCluster().getRegionServerThreads().stream()
315        .map(t -> t.getRegionServer().getServerName()).filter(sn -> !sn.equals(serverName))
316        .findAny().get();
317    Admin admin = TEST_UTIL.getAdmin();
318    RegionInfo region = admin.getRegions(TABLE_NAME).stream().findAny().get();
319    admin.move(region.getEncodedNameAsBytes(), Bytes.toBytes(newServerName.getServerName()));
320    TEST_UTIL.waitFor(30000, new ExplainingPredicate<Exception>() {
321
322      @Override
323      public boolean evaluate() throws Exception {
324        ServerName newServerName = TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME).getServerName();
325        return newServerName != null && !newServerName.equals(serverName);
326      }
327
328      @Override
329      public String explainFailure() throws Exception {
330        return region.getRegionNameAsString() + " is still on " + serverName;
331      }
332
333    });
334    // The cached location will not change
335    for (RegionLocateType locateType : RegionLocateType.values()) {
336      assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, serverName,
337        LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, locateType, false).get());
338    }
339    // should get the new location when reload = true
340    assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, newServerName,
341      LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, RegionLocateType.CURRENT, true).get());
342    // the cached location should be replaced
343    for (RegionLocateType locateType : RegionLocateType.values()) {
344      assertLocEquals(EMPTY_START_ROW, EMPTY_END_ROW, newServerName,
345        LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_START_ROW, locateType, false).get());
346    }
347  }
348
349  // Testcase for HBASE-20822
350  @Test
351  public void testLocateBeforeLastRegion()
352      throws IOException, InterruptedException, ExecutionException {
353    createMultiRegionTable();
354    LOCATOR.getRegionLocation(TABLE_NAME, SPLIT_KEYS[0], RegionLocateType.CURRENT, false).join();
355    HRegionLocation loc =
356      LOCATOR.getRegionLocation(TABLE_NAME, EMPTY_END_ROW, RegionLocateType.BEFORE, false).get();
357    // should locate to the last region
358    assertArrayEquals(loc.getRegion().getEndKey(), EMPTY_END_ROW);
359  }
360}