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