HBASE-23117: Bad enum in hbase:meta info:state column can fail loadMeta and stop startup (#867)

* Handling the BAD value in info:state columns in hbase:meta

* Adding a unit test and region encoded name in the LOG

* Adding a null check for region state to complete the test scenario and fixing the nit

Signed-off-by: Wellington Chevreuil <wchevreuil@apache.org>
Signed-off-by: GuangxuCheng  <guangxucheng@gmail.com>
Signed-off-by: stack <stack@apache.org>

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStateStore.java

@@ -132,7 +132,7 @@ public class RegionStateStore {
       if (regionInfo == null) continue;
 
       final int replicaId = regionInfo.getReplicaId();
-      final State state = getRegionState(result, replicaId);
+      final State state = getRegionState(result, replicaId, regionInfo);
 
       final ServerName lastHost = hrl.getServerName();
       final ServerName regionLocation = getRegionServer(result, replicaId);
@@ -347,13 +347,22 @@ public class RegionStateStore {
    * @return the region state, or null if unknown.
    */
   @VisibleForTesting
-  public static State getRegionState(final Result r, int replicaId) {
+  public static State getRegionState(final Result r, int replicaId, RegionInfo regionInfo) {
     Cell cell = r.getColumnLatestCell(HConstants.CATALOG_FAMILY, getStateColumn(replicaId));
     if (cell == null || cell.getValueLength() == 0) {
       return null;
     }
-    return State.valueOf(Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
+
-        cell.getValueLength()));
+    String state = Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
+        cell.getValueLength());
+    try {
+      return State.valueOf(state);
+    } catch (IllegalArgumentException e) {
+      LOG.warn("BAD value {} in hbase:meta info:state column for region {} , " +
+              "Consider using HBCK2 setRegionState ENCODED_REGION_NAME STATE",
+          state, regionInfo.getEncodedName());
+      return null;
+    }
   }
 
   private static byte[] getStateColumn(int replicaId) {

hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java

@@ -3606,7 +3606,7 @@ public class HBaseTestingUtility extends HBaseZKTestingUtility {
                     }
                   }
                   if (RegionStateStore.getRegionState(r,
-                    info.getReplicaId()) != RegionState.State.OPEN) {
+                    info.getReplicaId(), info) != RegionState.State.OPEN) {
                     return false;
                   }
                 }

hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRegionStateStore.java

@@ -19,20 +19,26 @@ package org.apache.hadoop.hbase.master.assignment;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
 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.Result;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.master.RegionState;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.testclassification.MasterTests;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.ClassRule;
@@ -84,6 +90,39 @@ public class TestRegionStateStore {
     assertTrue("Visitor has not been called.", visitorCalled.get());
   }
 
+  @Test
+  public void testVisitMetaForBadRegionState() throws Exception {
+    final TableName tableName = TableName.valueOf("testVisitMetaForBadRegionState");
+    util.createTable(tableName, "cf");
+    final List<HRegion> regions = util.getHBaseCluster().getRegions(tableName);
+    final String encodedName = regions.get(0).getRegionInfo().getEncodedName();
+    final RegionStateStore regionStateStore = util.getHBaseCluster().getMaster().
+        getAssignmentManager().getRegionStateStore();
+
+    // add the BAD_STATE which does not exist in enum RegionState.State
+    Put put = new Put(regions.get(0).getRegionInfo().getRegionName(),
+        EnvironmentEdgeManager.currentTime());
+    put.addColumn(HConstants.CATALOG_FAMILY, HConstants.STATE_QUALIFIER,
+        Bytes.toBytes("BAD_STATE"));
+
+    try (Table table = util.getConnection().getTable(TableName.META_TABLE_NAME)) {
+      table.put(put);
+    }
+
+    final AtomicBoolean visitorCalled = new AtomicBoolean(false);
+    regionStateStore.visitMetaForRegion(encodedName, new RegionStateStore.RegionStateVisitor() {
+      @Override
+      public void visitRegionState(Result result, RegionInfo regionInfo,
+                                   RegionState.State state, ServerName regionLocation,
+                                   ServerName lastHost, long openSeqNum) {
+        assertEquals(encodedName, regionInfo.getEncodedName());
+        assertNull(state);
+        visitorCalled.set(true);
+      }
+    });
+    assertTrue("Visitor has not been called.", visitorCalled.get());
+  }
+
   @Test
   public void testVisitMetaForRegionNonExistingRegion() throws Exception {
     final String encodedName = "fakeencodedregionname";