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