HBASE-4334 HRegion.get never validates row (Lars H)

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1178623 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -330,6 +330,7 @@ Release 0.92.0 - Unreleased
    HBASE-4496  HFile V2 does not honor setCacheBlocks when scanning (Lars and Mikhail)
    HBASE-4531  hbase-4454 failsafe broke mvn site; back it out or fix
                (Akash Ashok)
+   HBASE-4334  HRegion.get never validates row (Lars Hofhansl)
 
   TESTS
    HBASE-4450  test for number of blocks read: to serve as baseline for expected

src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -1335,7 +1335,8 @@ public class HRegion implements HeapSize { // , Writable{
     }
   }
 
-  protected RegionScanner getScanner(Scan scan, List<KeyValueScanner> additionalScanners) throws IOException {
+  protected RegionScanner getScanner(Scan scan,
+      List<KeyValueScanner> additionalScanners) throws IOException {
     startRegionOperation();
     this.readRequestsCount.increment();
     try {
@@ -1347,7 +1348,6 @@ public class HRegion implements HeapSize { // , Writable{
         }
       }
       return instantiateRegionScanner(scan, additionalScanners);
-
     } finally {
       closeRegionOperation();
     }
@@ -2427,10 +2427,10 @@ public class HRegion implements HeapSize { // , Writable{
   //////////////////////////////////////////////////////////////////////////////
 
   /** Make sure this is a valid row for the HRegion */
-  private void checkRow(final byte [] row, String op) throws IOException {
+  void checkRow(final byte [] row, String op) throws IOException {
     if(!rowIsInRange(regionInfo, row)) {
       throw new WrongRegionException("Requested row out of range for " +
-          op + "on HRegion " + this + ", startKey='" +
+          op + " on HRegion " + this + ", startKey='" +
           Bytes.toStringBinary(regionInfo.getStartKey()) + "', getEndKey()='" +
           Bytes.toStringBinary(regionInfo.getEndKey()) + "', row='" +
           Bytes.toStringBinary(row) + "'");
@@ -3369,6 +3369,7 @@ public class HRegion implements HeapSize { // , Writable{
    * @throws IOException read exceptions
    */
   public Result get(final Get get, final Integer lockid) throws IOException {
+    checkRow(get.getRow(), "Get");
     // Verify families are all valid
     if (get.hasFamilies()) {
       for (byte [] family: get.familySet()) {

src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -2038,6 +2038,7 @@ public class HRegionServer implements HRegionInterface, HBaseRPCErrorHandler,
     requestCount.incrementAndGet();
     try {
       HRegion r = getRegion(regionName);
+      r.checkRow(scan.getStartRow(), "Scan");
       r.prepareScanner(scan);
       RegionScanner s = null;
       if (r.getCoprocessorHost() != null) {

src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java

@@ -2792,6 +2792,24 @@ public class TestHRegion extends HBaseTestCase {
     flushThread.checkNoError();
   }
 
+  public void testHolesInMeta() throws Exception {
+    String method = "testHolesInMeta";
+    byte[] tableName = Bytes.toBytes(method);
+    byte[] family = Bytes.toBytes("family");
+    initHRegion(tableName, Bytes.toBytes("x"), Bytes.toBytes("z"), method,
+        HBaseConfiguration.create(), family);
+    byte[] rowNotServed = Bytes.toBytes("a");
+    Get g = new Get(rowNotServed);
+    try {
+      region.get(g, null);
+      fail();
+    } catch (WrongRegionException x) {
+      // OK
+    }
+    byte[] row = Bytes.toBytes("y");
+    g = new Get(row);
+    region.get(g, null);
+  }
 
   public void testIndexesScanWithOneDeletedRow() throws IOException {
     byte[] tableName = Bytes.toBytes("testIndexesScanWithOneDeletedRow");
@@ -3142,13 +3160,19 @@ public class TestHRegion extends HBaseTestCase {
   }
 
   private void initHRegion (byte [] tableName, String callingMethod,
-    Configuration conf, byte [] ... families)
-  throws IOException{
+      Configuration conf, byte [] ... families)
+    throws IOException{
+    initHRegion(tableName, null, null, callingMethod, conf, families);
+  }
+
+  private void initHRegion(byte[] tableName, byte[] startKey, byte[] stopKey,
+      String callingMethod, Configuration conf, byte[]... families)
+      throws IOException {
     HTableDescriptor htd = new HTableDescriptor(tableName);
     for(byte [] family : families) {
       htd.addFamily(new HColumnDescriptor(family));
     }
-    HRegionInfo info = new HRegionInfo(htd.getName(), null, null, false);
+    HRegionInfo info = new HRegionInfo(htd.getName(), startKey, stopKey, false);
     Path path = new Path(DIR + callingMethod);
     if (fs.exists(path)) {
       if (!fs.delete(path, true)) {