From 3dd220f8fdf6e34e11ffa309ff182172f6346091 Mon Sep 17 00:00:00 2001
From: tedyu <yuzhihong@gmail.com>
Date: Wed, 11 Feb 2015 12:45:54 -0800
Subject: [PATCH] HBASE-13018 WALSplitter should not try to get table states
 while splitting META (Andrey Stepachev)

---
 .../apache/hadoop/hbase/wal/WALSplitter.java  | 37 ++++++++++++-------
 .../master/TestDistributedLogSplitting.java   |  1 +
 2 files changed, 25 insertions(+), 13 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
index c187af1fb84..4d8cc2dcc2b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java
@@ -65,6 +65,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.MetaTableAccessor;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.TableNotFoundException;
@@ -146,8 +147,8 @@ public class WALSplitter {
   OutputSink outputSink;
   EntryBuffers entryBuffers;
 
-  private Set<TableName> disablingOrDisabledTables =
-      new HashSet<TableName>();
+  private Map<TableName, TableState> tableStatesCache =
+      new ConcurrentHashMap<>();
   private BaseCoordinatedStateManager csm;
   private final WALFactory walFactory;
 
@@ -303,16 +304,6 @@ public class WALSplitter {
         LOG.warn("Nothing to split in log file " + logPath);
         return true;
       }
-      if(csm != null) {
-        HConnection scc = csm.getServer().getConnection();
-        TableName[] tables = scc.listTableNames();
-        for (TableName table : tables) {
-          if (scc.getTableState(table)
-              .inStates(TableState.State.DISABLED, TableState.State.DISABLING)) {
-            disablingOrDisabledTables.add(table);
-          }
-        }
-      }
       int numOpenedFilesBeforeReporting = conf.getInt("hbase.splitlog.report.openedfiles", 3);
       int numOpenedFilesLastCheck = 0;
       outputSink.setReporter(reporter);
@@ -1625,7 +1616,7 @@ public class WALSplitter {
       }
 
       // check if current region in a disabling or disabled table
-      if (disablingOrDisabledTables.contains(buffer.tableName)) {
+      if (isTableDisabledOrDisabling(buffer.tableName)) {
         // need fall back to old way
         logRecoveredEditsOutputSink.append(buffer);
         hasEditsInDisablingOrDisabledTables = true;
@@ -2057,6 +2048,26 @@ public class WALSplitter {
       return this.recoveredRegions.size();
     }
 
+    private boolean isTableDisabledOrDisabling(TableName tableName) {
+      if (csm == null)
+        return false; // we can't get state without CoordinatedStateManager
+      if (tableName.isSystemTable())
+        return false; // assume that system tables never can be disabled
+      TableState tableState = tableStatesCache.get(tableName);
+      if (tableState == null) {
+        try {
+          tableState =
+              MetaTableAccessor.getTableState(csm.getServer().getConnection(), tableName);
+          if (tableState != null)
+            tableStatesCache.put(tableName, tableState);
+        } catch (IOException e) {
+          LOG.warn("State is not accessible for table " + tableName, e);
+        }
+      }
+      return tableState != null && tableState
+          .inStates(TableState.State.DISABLED, TableState.State.DISABLING);
+    }
+
     /**
      * Get a writer and path for a log starting at the given entry. This function is threadsafe so
      * long as multiple threads are always acting on different regions.
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestDistributedLogSplitting.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestDistributedLogSplitting.java
index f2376318fbb..0038d71befa 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestDistributedLogSplitting.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestDistributedLogSplitting.java
@@ -807,6 +807,7 @@ public class TestDistributedLogSplitting {
 
     LOG.info("Disabling table\n");
     TEST_UTIL.getHBaseAdmin().disableTable(TableName.valueOf("disableTable"));
+    TEST_UTIL.waitTableDisabled(TableName.valueOf("disableTable").getName());
 
     // abort RS
     LOG.info("Aborting region server: " + hrs.getServerName());