Checking replica id on AsyncRpcRetryingCaller side

guluo2016 · guluo2016 · commit 506935a7317b · 2024-03-26T23:14:48.000+08:00
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRegionLocator.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRegionLocator.java
@@ -35,7 +35,6 @@
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.RegionLocations;
 import org.apache.hadoop.hbase.ServerName;
@@ -152,67 +151,36 @@ CompletableFuture<RegionLocations> getRegionLocations(TableName tableName, byte[
     }, AsyncRegionLocator::getRegionNames, supplier);
   }
 
-  private void internalAddListener(CompletableFuture<HRegionLocation> future,
-    CompletableFuture<RegionLocations> locsFuture, TableName tableName, byte[] row, int replicaId,
-    RegionLocateType type) {
-    addListener(locsFuture, (locs, error) -> {
-      if (error != null) {
-        future.completeExceptionally(error);
-        return;
-      }
-      HRegionLocation loc = locs.getRegionLocation(replicaId);
-      if (loc == null) {
-        future.completeExceptionally(
-          new RegionOfflineException("No location for " + tableName + ", row='"
-            + Bytes.toStringBinary(row) + "', locateType=" + type + ", replicaId=" + replicaId));
-      } else if (loc.getServerName() == null) {
-        future
-          .completeExceptionally(new RegionOfflineException("No server address listed for region '"
-            + loc.getRegion().getRegionNameAsString() + ", row='" + Bytes.toStringBinary(row)
-            + "', locateType=" + type + ", replicaId=" + replicaId));
-      } else {
-        future.complete(loc);
-      }
-    });
-  }
-
   CompletableFuture<HRegionLocation> getRegionLocation(TableName tableName, byte[] row,
     int replicaId, RegionLocateType type, boolean reload, long timeoutNs) {
     final Supplier<Span> supplier = new TableSpanBuilder(conn)
       .setName("AsyncRegionLocator.getRegionLocation").setTableName(tableName);
     return tracedLocationFuture(() -> {
+      // meta region can not be split right now so we always call the same method.
+      // Change it later if the meta table can have more than one regions.
       CompletableFuture<HRegionLocation> future = new CompletableFuture<>();
-      if (replicaId == RegionReplicaUtil.DEFAULT_REPLICA_ID) {
-        // meta region can not be split right now so we always call the same method.
-        // Change it later if the meta table can have more than one regions.
-        CompletableFuture<RegionLocations> locsFuture = isMeta(tableName)
-          ? metaRegionLocator.getRegionLocations(replicaId, reload)
-          : nonMetaRegionLocator.getRegionLocations(tableName, row, replicaId, type, reload);
-        internalAddListener(future, locsFuture, tableName, row, replicaId, type);
-      } else {
-        addListener(conn.getAdmin().getDescriptor(tableName), (tdesc, error) -> {
-          if (error != null) {
-            future.completeExceptionally(error);
-            return;
-          }
-          int regionReplicationCount = tdesc.getRegionReplication();
-          if (replicaId >= regionReplicationCount) {
-            future
-              .completeExceptionally(new DoNotRetryIOException("The specified region replica id "
-                + replicaId + " does not exist, the REGION_REPLICATION of this table "
-                + tableName.getNameAsString() + " is " + regionReplicationCount + ","
-                + " this means that the maximum region replica id you can specify is "
-                + (regionReplicationCount - 1) + "."));
-            return;
-          }
-          // meta region can not be split right now so we always call the same method.
-          // Change it later if the meta table can have more than one regions.
-          CompletableFuture<RegionLocations> locsFuture = isMeta(tableName)
-            ? metaRegionLocator.getRegionLocations(replicaId, reload)
-            : nonMetaRegionLocator.getRegionLocations(tableName, row, replicaId, type, reload);
-          internalAddListener(future, locsFuture, tableName, row, replicaId, type);
-        });
-      }
+      CompletableFuture<RegionLocations> locsFuture = isMeta(tableName)
+        ? metaRegionLocator.getRegionLocations(replicaId, reload)
+        : nonMetaRegionLocator.getRegionLocations(tableName, row, replicaId, type, reload);
+      addListener(locsFuture, (locs, error) -> {
+        if (error != null) {
+          future.completeExceptionally(error);
+          return;
+        }
+        HRegionLocation loc = locs.getRegionLocation(replicaId);
+        if (loc == null) {
+          future.completeExceptionally(
+            new RegionOfflineException("No location for " + tableName + ", row='"
+              + Bytes.toStringBinary(row) + "', locateType=" + type + ", replicaId=" + replicaId));
+        } else if (loc.getServerName() == null) {
+          future.completeExceptionally(
+            new RegionOfflineException("No server address listed for region '"
+              + loc.getRegion().getRegionNameAsString() + ", row='" + Bytes.toStringBinary(row)
+              + "', locateType=" + type + ", replicaId=" + replicaId));
+        } else {
+          future.complete(loc);
+        }
+      });
       return withTimeout(future, timeoutNs,
         () -> "Timeout(" + TimeUnit.NANOSECONDS.toMillis(timeoutNs)
           + "ms) waiting for region location for " + tableName + ", row='"
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRpcRetryingCaller.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRpcRetryingCaller.java
@@ -204,6 +204,29 @@ protected final void onError(Throwable t, Supplier<String> errMsg,
           }
           if (disabled) {
             future.completeExceptionally(new TableNotEnabledException(tableName.get()));
+            return;
+          }
+          if (this instanceof AsyncSingleRequestRpcRetryingCaller) {
+            AsyncSingleRequestRpcRetryingCaller caller = (AsyncSingleRequestRpcRetryingCaller) this;
+            int replicaId = caller.getReplicaId();
+            FutureUtils.addListener(conn.getAdmin().getDescriptor(tableName.get()),
+              (tdesc, tdescError) -> {
+                if (tdescError != null) {
+                  future.completeExceptionally(tdescError);
+                  return;
+                }
+                int regionReplicationCount = tdesc.getRegionReplication();
+                if (replicaId >= regionReplicationCount) {
+                  future.completeExceptionally(
+                    new DoNotRetryIOException("The specified region replica id " + replicaId
+                      + " does not exist, the REGION_REPLICATION of this table "
+                      + tableName.get().getNameAsString() + " is " + regionReplicationCount + ","
+                      + " this means that the maximum region replica id you can specify is "
+                      + (regionReplicationCount - 1) + "."));
+                  return;
+                }
+                tryScheduleRetry(error);
+              });
           } else {
             tryScheduleRetry(error);
           }
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncSingleRequestRpcRetryingCaller.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncSingleRequestRpcRetryingCaller.java
@@ -122,4 +122,8 @@ protected void doCall() {
   protected Optional<TableName> getTableName() {
     return Optional.of(tableName);
   }
+
+  public int getReplicaId() {
+    return replicaId;
+  }
 }
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScanOrGetWithReplicationFromClient.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestScanOrGetWithReplicationFromClient.java
@@ -18,17 +18,15 @@
 package org.apache.hadoop.hbase.client;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
+import java.io.IOException;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtil;
-import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.filter.PrefixFilter;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.testclassification.ClientTests;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
@@ -46,10 +44,10 @@
 import org.junit.rules.TestName;
 
 @Category({ MediumTests.class, ClientTests.class })
-public class TestAsyncTableRegionLocatorWithRegionReplicaId {
+public class TestScanOrGetWithReplicationFromClient {
   @ClassRule
   public static final HBaseClassTestRule CLASS_RULE =
-    HBaseClassTestRule.forClass(TestAsyncTableRegionLocatorWithRegionReplicaId.class);
+    HBaseClassTestRule.forClass(TestScanOrGetWithReplicationFromClient.class);
 
   @Rule
   public TestName name = new TestName();
@@ -63,15 +61,13 @@ public class TestAsyncTableRegionLocatorWithRegionReplicaId {
   // region replica id starts from 0
   private static final int NON_EXISTING_REGION_REPLICA_ID = REGION_REPLICATION_COUNT;
   private static Connection connection;
-  private static AsyncConnection asyncConn;
   private static Admin admin;
   private TableName tableName;
 
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
     UTIL.startMiniCluster(1);
     connection = UTIL.getConnection();
-    asyncConn = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get();
     admin = UTIL.getAdmin();
   }
 
@@ -115,60 +111,98 @@ public void tearDown() throws Exception {
   }
 
   @Test
-  public void testMetaTableRegionLocatorWithRegionReplicaId()
-    throws ExecutionException, InterruptedException {
-    AsyncTableRegionLocator locator = asyncConn.getRegionLocator(TableName.META_TABLE_NAME);
-    CompletableFuture<HRegionLocation> future =
-      locator.getRegionLocation(tableName.getName(), RegionReplicaUtil.DEFAULT_REPLICA_ID, true);
-    HRegionLocation hrl = future.get();
-    assertNotNull(hrl);
+  public void testScanMetaWithRegionReplicaId() throws IOException {
+    Table metaTable = connection.getTable(TableName.META_TABLE_NAME);
+    Scan scan = new Scan();
+    scan.setFilter(new PrefixFilter(tableName.getName()));
+    scan.setReplicaId(RegionReplicaUtil.DEFAULT_REPLICA_ID);
+    scan.setConsistency(Consistency.TIMELINE);
+    ResultScanner rs = metaTable.getScanner(scan);
+    rs.forEach(r -> assertTrue(Bytes.toString(r.getRow()).contains(tableName.getNameAsString())));
   }
 
   @Test
-  public void testMetaTableRegionLocatorWithNonExistingRegionReplicaId()
-    throws InterruptedException {
-    AsyncTableRegionLocator locator = asyncConn.getRegionLocator(TableName.META_TABLE_NAME);
-    CompletableFuture<HRegionLocation> future =
-      locator.getRegionLocation(tableName.getName(), NON_EXISTING_REGION_REPLICA_ID, true);
+  public void testScanMetaWithNonExistingRegionReplicaId() throws IOException {
+    Table metaTable = connection.getTable(TableName.META_TABLE_NAME);
+    Scan scan = new Scan();
+    scan.setReplicaId(NON_EXISTING_REGION_REPLICA_ID);
+    scan.setConsistency(Consistency.TIMELINE);
+    exception.expect(DoNotRetryIOException.class);
+    ResultScanner rs = metaTable.getScanner(scan);
     try {
-      future.get();
-    } catch (ExecutionException e) {
-      assertTrue(e.getCause() instanceof DoNotRetryIOException);
+      rs.forEach(r -> Bytes.toString(r.getRow()));
+    } catch (Exception e) {
+      Throwable throwable = e.getCause();
+      assertTrue(throwable instanceof DoNotRetryIOException);
       String message = "The specified region replica id " + NON_EXISTING_REGION_REPLICA_ID
         + " does not exist, the REGION_REPLICATION of this table "
         + TableName.META_TABLE_NAME.getNameAsString() + " is "
         + TableDescriptorBuilder.DEFAULT_REGION_REPLICATION + ", "
         + "this means that the maximum region replica id you can specify is "
         + (TableDescriptorBuilder.DEFAULT_REGION_REPLICATION - 1) + ".";
-      assertEquals(message, e.getCause().getMessage());
+      assertEquals(message, throwable.getMessage());
+    }
+  }
+
+  @Test
+  public void testScanTableWithRegionReplicaId() throws IOException {
+    Table table = connection.getTable(tableName);
+    Scan scan = new Scan();
+    scan.setReplicaId(RegionReplicaUtil.DEFAULT_REPLICA_ID);
+    scan.setConsistency(Consistency.TIMELINE);
+    ResultScanner rs = table.getScanner(scan);
+    rs.forEach(r -> assertEquals(ROW, Bytes.toString(r.getRow())));
+  }
+
+  @Test
+  public void testScanTableWithNonExistingRegionReplicaId() throws IOException {
+    Table table = connection.getTable(tableName);
+    Scan scan = new Scan();
+    scan.setReplicaId(NON_EXISTING_REGION_REPLICA_ID);
+    scan.setConsistency(Consistency.TIMELINE);
+    exception.expect(DoNotRetryIOException.class);
+    ResultScanner rs = table.getScanner(scan);
+    try {
+      rs.forEach(r -> Bytes.toString(r.getRow()));
+    } catch (Exception e) {
+      Throwable throwable = e.getCause();
+      assertTrue(throwable instanceof DoNotRetryIOException);
+      String message = "The specified region replica id " + NON_EXISTING_REGION_REPLICA_ID
+        + " does not exist, the REGION_REPLICATION of this table " + tableName.getNameAsString()
+        + " is " + REGION_REPLICATION_COUNT + ", "
+        + "this means that the maximum region replica id you can specify is "
+        + (REGION_REPLICATION_COUNT - 1) + ".";
+      assertEquals(message, throwable.getMessage());
     }
   }
 
   @Test
-  public void testTableRegionLocatorWithRegionReplicaId()
-    throws ExecutionException, InterruptedException {
-    AsyncTableRegionLocator locator = asyncConn.getRegionLocator(tableName);
-    CompletableFuture<HRegionLocation> future =
-      locator.getRegionLocation(Bytes.toBytes(ROW), RegionReplicaUtil.DEFAULT_REPLICA_ID, true);
-    HRegionLocation hrl = future.get();
-    assertNotNull(hrl);
+  public void testGetTableWithRegionReplicaId() throws IOException {
+    Table table = connection.getTable(tableName);
+    Get get = new Get(Bytes.toBytes(ROW)).setConsistency(Consistency.TIMELINE)
+      .setReplicaId(RegionReplicaUtil.DEFAULT_REPLICA_ID);
+    Result result = table.get(get);
+    assertEquals(ROW, Bytes.toString(result.getRow()));
+    String value = Bytes.toString(result.getValue(FAMILY, Bytes.toBytes("q")));
+    assertEquals("test_value", value);
   }
 
   @Test
-  public void testTableRegionLocatorWithNonExistingRegionReplicaId() throws InterruptedException {
-    AsyncTableRegionLocator locator = asyncConn.getRegionLocator(tableName);
-    CompletableFuture<HRegionLocation> future =
-      locator.getRegionLocation(Bytes.toBytes(ROW), NON_EXISTING_REGION_REPLICA_ID, true);
+  public void testGetTableWithNonExistingRegionReplicaId() throws IOException {
+    Table table = connection.getTable(tableName);
+    Get get = new Get(Bytes.toBytes(ROW)).setConsistency(Consistency.TIMELINE)
+      .setReplicaId(NON_EXISTING_REGION_REPLICA_ID);
     try {
-      future.get();
-    } catch (ExecutionException e) {
-      assertTrue(e.getCause() instanceof DoNotRetryIOException);
+      Result result = table.get(get);
+      result.getValue(FAMILY, Bytes.toBytes("q"));
+    } catch (Exception e) {
+      assertTrue(e instanceof DoNotRetryIOException);
       String message = "The specified region replica id " + NON_EXISTING_REGION_REPLICA_ID
         + " does not exist, the REGION_REPLICATION of this table " + tableName.getNameAsString()
         + " is " + REGION_REPLICATION_COUNT + ", "
         + "this means that the maximum region replica id you can specify is "
         + (REGION_REPLICATION_COUNT - 1) + ".";
-      assertEquals(message, e.getCause().getMessage());
+      assertEquals(message, e.getMessage());
     }
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -122,4 +122,8 @@ protected void doCall() {`
`122`	`122`	`protected Optional<TableName> getTableName() {`
`123`	`123`	`return Optional.of(tableName);`
`124`	`124`	`}`
	`125`	`+`
	`126`	`+ public int getReplicaId() {`
	`127`	`+ return replicaId;`
	`128`	`+ }`
`125`	`129`	`}`