HBASE-25272 Support scan on a specific replica (#2645)

Signed-off-by: stack <stack@apache.org>
Signed-off-by: Huaxiang Sun <huaxiangsun@apache.org>

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java

@@ -21,33 +21,32 @@ import static org.apache.hadoop.hbase.client.ConnectionUtils.calcEstimatedSize;
 import static org.apache.hadoop.hbase.client.ConnectionUtils.createScanResultCache;
 import static org.apache.hadoop.hbase.client.ConnectionUtils.incRegionCountMetrics;
 
-import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
-
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.ArrayDeque;
 import java.util.Queue;
 import java.util.concurrent.ExecutorService;
-
 import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.client.ScannerCallable.MoreResults;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
-import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
-import org.apache.hadoop.hbase.exceptions.ScannerResetException;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
-import org.apache.hadoop.hbase.regionserver.LeaseException;
 import org.apache.hadoop.hbase.NotServingRegionException;
-import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.UnknownScannerException;
+import org.apache.hadoop.hbase.client.ScannerCallable.MoreResults;
+import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
+import org.apache.hadoop.hbase.exceptions.ScannerResetException;
+import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
+import org.apache.hadoop.hbase.regionserver.LeaseException;
+import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
+
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 
 /**
@@ -142,6 +141,11 @@ public abstract class ClientScanner extends AbstractClientScanner {
     initCache();
   }
 
+  protected final int getScanReplicaId() {
+    return scan.getReplicaId() >= RegionReplicaUtil.DEFAULT_REPLICA_ID ? scan.getReplicaId() :
+      RegionReplicaUtil.DEFAULT_REPLICA_ID;
+  }
+
   protected ClusterConnection getConnection() {
     return this.connection;
   }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSimpleScanner.java

@@ -61,6 +61,6 @@ public class ClientSimpleScanner extends ClientScanner {
       scan.withStartRow(createClosestRowAfter(scan.getStartRow()), true);
     }
     return new ScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
-        this.rpcControllerFactory);
+      this.rpcControllerFactory, getScanReplicaId());
   }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ReversedClientScanner.java

@@ -37,13 +37,6 @@ public class ReversedClientScanner extends ClientScanner {
   /**
    * Create a new ReversibleClientScanner for the specified table Note that the passed
    * {@link Scan}'s start row maybe changed.
-   * @param conf
-   * @param scan
-   * @param tableName
-   * @param connection
-   * @param pool
-   * @param primaryOperationTimeout
-   * @throws IOException
    */
   public ReversedClientScanner(Configuration conf, Scan scan, TableName tableName,
       ClusterConnection connection, RpcRetryingCallerFactory rpcFactory,
@@ -65,6 +58,6 @@ public class ReversedClientScanner extends ClientScanner {
   @Override
   protected ReversedScannerCallable createScannerCallable() {
     return new ReversedScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
-        this.rpcControllerFactory);
+        this.rpcControllerFactory, getScanReplicaId());
   }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ReversedScannerCallable.java

@@ -45,35 +45,22 @@ import org.apache.hadoop.hbase.util.Bytes;
 public class ReversedScannerCallable extends ScannerCallable {
 
   /**
-   * @param connection
-   * @param tableName
-   * @param scan
-   * @param scanMetrics
-   * @param rpcFactory to create an {@link com.google.protobuf.RpcController} to talk to the
-   *          regionserver
-   */
-  public ReversedScannerCallable(ClusterConnection connection, TableName tableName, Scan scan,
-      ScanMetrics scanMetrics, RpcControllerFactory rpcFactory) {
-    super(connection, tableName, scan, scanMetrics, rpcFactory);
-  }
-
-  /**
-   * @param connection
-   * @param tableName
-   * @param scan
-   * @param scanMetrics
+   * @param connection which connection
+   * @param tableName table callable is on
+   * @param scan the scan to execute
+   * @param scanMetrics the ScanMetrics to used, if it is null, ScannerCallable won't collect
+   *          metrics
    * @param rpcFactory to create an {@link com.google.protobuf.RpcController} to talk to the
    *          regionserver
    * @param replicaId the replica id
    */
   public ReversedScannerCallable(ClusterConnection connection, TableName tableName, Scan scan,
-      ScanMetrics scanMetrics, RpcControllerFactory rpcFactory, int replicaId) {
+    ScanMetrics scanMetrics, RpcControllerFactory rpcFactory, int replicaId) {
     super(connection, tableName, scan, scanMetrics, rpcFactory, replicaId);
   }
 
   /**
    * @param reload force reload of server location
-   * @throws IOException
    */
   @Override
   public void prepare(boolean reload) throws IOException {
@@ -86,9 +73,8 @@ public class ReversedScannerCallable extends ScannerCallable {
       // 2. the start row is empty which means we need to locate to the last region.
       if (scan.includeStartRow() && !isEmptyStartRow(getRow())) {
         // Just locate the region with the row
-        RegionLocations rl = RpcRetryingCallerWithReadReplicas.getRegionLocations(!reload, id,
-            getConnection(), getTableName(), getRow());
-        this.location = id < rl.size() ? rl.getRegionLocation(id) : null;
+        RegionLocations rl = getRegionLocations(reload, getRow());
+        this.location = getLocationForReplica(rl);
         if (location == null || location.getServerName() == null) {
           throw new IOException("Failed to find location, tableName="
               + getTableName() + ", row=" + Bytes.toStringBinary(getRow()) + ", reload="
@@ -126,7 +112,6 @@ public class ReversedScannerCallable extends ScannerCallable {
    * @param reload force reload of server location
    * @return A list of HRegionLocation corresponding to the regions that contain
    *         the specified range
-   * @throws IOException
    */
   private List<HRegionLocation> locateRegionsInRange(byte[] startKey,
       byte[] endKey, boolean reload) throws IOException {
@@ -140,15 +125,14 @@ public class ReversedScannerCallable extends ScannerCallable {
     List<HRegionLocation> regionList = new ArrayList<>();
     byte[] currentKey = startKey;
     do {
-      RegionLocations rl = RpcRetryingCallerWithReadReplicas.getRegionLocations(!reload, id,
-          getConnection(), getTableName(), currentKey);
-      HRegionLocation regionLocation = id < rl.size() ? rl.getRegionLocation(id) : null;
-      if (regionLocation != null && regionLocation.getRegionInfo().containsRow(currentKey)) {
+      RegionLocations rl = getRegionLocations(reload, currentKey);
+      HRegionLocation regionLocation = getLocationForReplica(rl);
+      if (regionLocation.getRegionInfo().containsRow(currentKey)) {
         regionList.add(regionLocation);
       } else {
-        throw new DoNotRetryIOException("Does hbase:meta exist hole? Locating row "
-            + Bytes.toStringBinary(currentKey) + " returns incorrect region "
-            + (regionLocation == null ? null : regionLocation.getRegionInfo()));
+        throw new DoNotRetryIOException(
+          "Does hbase:meta exist hole? Locating row " + Bytes.toStringBinary(currentKey) +
+            " returns incorrect region " + regionLocation.getRegionInfo());
       }
       currentKey = regionLocation.getRegionInfo().getEndKey();
     } while (!Bytes.equals(currentKey, HConstants.EMPTY_END_ROW)

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java

@@ -26,7 +26,6 @@ import static org.apache.hadoop.hbase.client.ConnectionUtils.updateServerSideMet
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HBaseIOException;
@@ -37,13 +36,14 @@ import org.apache.hadoop.hbase.RegionLocations;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.UnknownScannerException;
-import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
 import org.apache.hadoop.hbase.exceptions.ScannerResetException;
 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
@@ -103,18 +103,6 @@ public class ScannerCallable extends ClientServiceCallable<Result[]> {
    * @param rpcControllerFactory factory to use when creating
    *        {@link com.google.protobuf.RpcController}
    */
-  public ScannerCallable(ClusterConnection connection, TableName tableName, Scan scan,
-      ScanMetrics scanMetrics, RpcControllerFactory rpcControllerFactory) {
-    this(connection, tableName, scan, scanMetrics, rpcControllerFactory, 0);
-  }
-  /**
-   *
-   * @param connection
-   * @param tableName
-   * @param scan
-   * @param scanMetrics
-   * @param id the replicaId
-   */
   public ScannerCallable(ClusterConnection connection, TableName tableName, Scan scan,
       ScanMetrics scanMetrics, RpcControllerFactory rpcControllerFactory, int id) {
     super(connection, tableName, scan.getStartRow(), rpcControllerFactory.newController(), scan.getPriority());
@@ -127,23 +115,33 @@ public class ScannerCallable extends ClientServiceCallable<Result[]> {
     this.rpcControllerFactory = rpcControllerFactory;
   }
 
+  protected final HRegionLocation getLocationForReplica(RegionLocations locs)
+    throws HBaseIOException {
+    HRegionLocation loc = id < locs.size() ? locs.getRegionLocation(id) : null;
+    if (loc == null || loc.getServerName() == null) {
+      // With this exception, there will be a retry. The location can be null for a replica
+      // when the table is created or after a split.
+      throw new HBaseIOException("There is no location for replica id #" + id);
+    }
+    return loc;
+  }
+
+  protected final RegionLocations getRegionLocations(boolean reload, byte[] row)
+    throws IOException {
+    return RpcRetryingCallerWithReadReplicas.getRegionLocations(!reload, id, getConnection(),
+      getTableName(), row);
+  }
+
   /**
    * @param reload force reload of server location
-   * @throws IOException
    */
   @Override
   public void prepare(boolean reload) throws IOException {
     if (Thread.interrupted()) {
       throw new InterruptedIOException();
     }
-    RegionLocations rl = RpcRetryingCallerWithReadReplicas.getRegionLocations(!reload,
-        id, getConnection(), getTableName(), getRow());
-    location = id < rl.size() ? rl.getRegionLocation(id) : null;
-    if (location == null || location.getServerName() == null) {
-      // With this exception, there will be a retry. The location can be null for a replica
-      //  when the table is created or after a split.
-      throw new HBaseIOException("There is no location for replica id #" + id);
-    }
+    RegionLocations rl = getRegionLocations(reload, getRow());
+    location = getLocationForReplica(rl);
     ServerName dest = location.getServerName();
     setStub(super.getConnection().getClient(dest));
     if (!instantiated || reload) {

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallableWithReplicas.java

@@ -127,7 +127,7 @@ class ScannerCallableWithReplicas implements RetryingCallable<Result[]> {
   }
 
   @Override
-  public Result [] call(int timeout) throws IOException {
+  public Result[] call(int timeout) throws IOException {
     // If the active replica callable was closed somewhere, invoke the RPC to
     // really close it. In the case of regular scanners, this applies. We make couple
     // of RPCs to a RegionServer, and when that region is exhausted, we set
@@ -187,7 +187,7 @@ class ScannerCallableWithReplicas implements RetryingCallable<Result[]> {
 
     AtomicBoolean done = new AtomicBoolean(false);
     replicaSwitched.set(false);
-    // submit call for the primary replica.
+    // submit call for the primary replica or user specified replica
     addCallsForCurrentReplica(cs);
     int startIndex = 0;
 
@@ -209,13 +209,13 @@ class ScannerCallableWithReplicas implements RetryingCallable<Result[]> {
         LOG.debug("Scan with primary region returns " + e.getCause());
       }
 
-      // If rl's size is 1 or scan's consitency is strong, it needs to throw
-      // out the exception from the primary replica
-      if ((regionReplication == 1) || (scan.getConsistency() == Consistency.STRONG)) {
+      // If rl's size is 1 or scan's consitency is strong, or scan is over specific replica,
+      // it needs to throw out the exception from the primary replica
+      if (regionReplication == 1 || scan.getConsistency() == Consistency.STRONG ||
+        scan.getReplicaId() >= 0) {
         // Rethrow the first exception
         RpcRetryingCallerWithReadReplicas.throwEnrichedException(e, retries);
       }
-
       startIndex = 1;
     } catch (CancellationException e) {
       throw new InterruptedIOException(e.getMessage());
@@ -225,8 +225,9 @@ class ScannerCallableWithReplicas implements RetryingCallable<Result[]> {
 
     // submit call for the all of the secondaries at once
     int endIndex = regionReplication;
-    if (scan.getConsistency() == Consistency.STRONG) {
-      // When scan's consistency is strong, do not send to the secondaries
+    if (scan.getConsistency() == Consistency.STRONG || scan.getReplicaId() >= 0) {
+      // When scan's consistency is strong or scan is over specific replica region, do not send to
+      // the secondaries
       endIndex = 1;
     } else {
       // TODO: this may be an overkill for large region replication

hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestReversedScannerCallable.java

@@ -75,7 +75,7 @@ public class TestReversedScannerCallable {
     Mockito.when(connection.relocateRegion(tableName, ROW, 0)).thenReturn(regionLocations);
 
     ReversedScannerCallable callable =
-        new ReversedScannerCallable(connection, tableName, scan, null, rpcFactory);
+      new ReversedScannerCallable(connection, tableName, scan, null, rpcFactory, 0);
     callable.prepare(true);
 
     Mockito.verify(connection).relocateRegion(tableName, ROW, 0);
@@ -85,10 +85,10 @@ public class TestReversedScannerCallable {
   public void testPrepareUsesCache() throws Exception {
     TableName tableName = TableName.valueOf("MyTable");
     Mockito.when(connection.locateRegion(tableName, ROW, true, true, 0))
-        .thenReturn(regionLocations);
+      .thenReturn(regionLocations);
 
     ReversedScannerCallable callable =
-        new ReversedScannerCallable(connection, tableName, scan, null, rpcFactory);
+      new ReversedScannerCallable(connection, tableName, scan, null, rpcFactory, 0);
     callable.prepare(false);
 
     Mockito.verify(connection).locateRegion(tableName, ROW, true, true, 0);

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerNoMaster.java

@@ -26,6 +26,7 @@ import org.apache.hadoop.hbase.NotServingRegionException;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.RegionLocator;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.TableDescriptor;
@@ -129,8 +130,11 @@ public class TestRegionServerNoMaster {
     }
   }
 
-  /** Flush the given region in the mini cluster. Since no master, we cannot use HBaseAdmin.flush() */
-  public static void flushRegion(HBaseTestingUtility HTU, HRegionInfo regionInfo) throws IOException {
+  /**
+   * Flush the given region in the mini cluster. Since no master, we cannot use HBaseAdmin.flush()
+   */
+  public static void flushRegion(HBaseTestingUtility HTU, RegionInfo regionInfo)
+    throws IOException {
     for (RegionServerThread rst : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
       HRegion region = rst.getRegionServer().getRegionByEncodedName(regionInfo.getEncodedName());
       if (region != null) {