HBASE-27091 Speed up the loading of table descriptor from filesystem (#4493)

Signed-off-by: Huaxiang Sun <huaxiangsun@apache.org>
Signed-off-by: Yu Li <liyu@apache.org>

hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java

@@ -231,8 +231,10 @@ public abstract class HBaseServerBase<R extends HBaseRpcServicesBase<?>> extends
     // init the filesystem
     this.dataFs = new HFileSystem(this.conf, useHBaseChecksum);
     this.dataRootDir = CommonFSUtils.getRootDir(this.conf);
+    int tableDescriptorParallelLoadThreads =
+      conf.getInt("hbase.tabledescriptor.parallel.load.threads", 0);
     this.tableDescriptors = new FSTableDescriptors(this.dataFs, this.dataRootDir,
-      !canUpdateTableDescriptor(), cacheTableDescriptor());
+      !canUpdateTableDescriptor(), cacheTableDescriptor(), tableDescriptorParallelLoadThreads);
   }
 
   public HBaseServerBase(Configuration conf, String name) throws IOException {
@@ -466,6 +468,17 @@ public abstract class HBaseServerBase<R extends HBaseRpcServicesBase<?>> extends
     }
   }
 
+  protected final void closeTableDescriptors() {
+    if (this.tableDescriptors != null) {
+      LOG.info("Close table descriptors");
+      try {
+        this.tableDescriptors.close();
+      } catch (IOException e) {
+        LOG.debug("Failed to close table descriptors gracefully", e);
+      }
+    }
+  }
+
   /**
    * In order to register ShutdownHook, this method is called when HMaster and HRegionServer are
    * started. For details, please refer to HBASE-26951

hbase-server/src/main/java/org/apache/hadoop/hbase/TableDescriptors.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.hadoop.hbase.client.TableDescriptor;
@@ -26,7 +27,7 @@ import org.apache.yetus.audience.InterfaceAudience;
  * Get, remove and modify table descriptors.
  */
 @InterfaceAudience.Private
-public interface TableDescriptors {
+public interface TableDescriptors extends Closeable {
 
   /**
    * Test whether a given table exists, i.e, has a table descriptor.
@@ -35,6 +36,11 @@ public interface TableDescriptors {
     return get(tableName) != null;
   }
 
+  @Override
+  default void close() throws IOException {
+    // do nothing by default
+  }
+
   /**
    * @return TableDescriptor for tablename
    */

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

@@ -601,6 +601,7 @@ public class HMaster extends HBaseServerBase<MasterRpcServices> implements Maste
           this.rpcServices.stop();
         }
         closeZooKeeper();
+        closeTableDescriptors();
         span.setStatus(StatusCode.OK);
       } finally {
         span.end();

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

@@ -977,6 +977,7 @@ public class HRegionServer extends HBaseServerBase<RSRpcServices>
       ZNodeClearer.deleteMyEphemeralNodeOnDisk();
 
       closeZooKeeper();
+      closeTableDescriptors();
       LOG.info("Exiting; stopping=" + this.serverName + "; zookeeper connection closed.");
       span.setStatus(StatusCode.OK);
     } finally {

hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSTableDescriptors.java

@@ -21,6 +21,7 @@ import com.google.errorprone.annotations.RestrictedApi;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.io.EOFException;
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
@@ -28,6 +29,12 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
@@ -55,6 +62,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.hbase.thirdparty.com.google.common.primitives.Ints;
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 /**
  * Implementation of {@link TableDescriptors} that reads descriptors from the passed filesystem. It
@@ -79,6 +87,8 @@ public class FSTableDescriptors implements TableDescriptors {
   private final boolean fsreadonly;
   private final boolean usecache;
   private volatile boolean fsvisited;
+  private boolean tableDescriptorParallelLoadEnable = false;
+  private ThreadPoolExecutor executor;
 
   long cachehits = 0;
   long invocations = 0;
@@ -108,10 +118,23 @@ public class FSTableDescriptors implements TableDescriptors {
 
   public FSTableDescriptors(final FileSystem fs, final Path rootdir, final boolean fsreadonly,
     final boolean usecache) {
+    this(fs, rootdir, fsreadonly, usecache, 0);
+  }
+
+  public FSTableDescriptors(final FileSystem fs, final Path rootdir, final boolean fsreadonly,
+    final boolean usecache, final int tableDescriptorParallelLoadThreads) {
     this.fs = fs;
     this.rootdir = rootdir;
     this.fsreadonly = fsreadonly;
     this.usecache = usecache;
+    if (tableDescriptorParallelLoadThreads > 0) {
+      tableDescriptorParallelLoadEnable = true;
+      executor = new ThreadPoolExecutor(tableDescriptorParallelLoadThreads,
+        tableDescriptorParallelLoadThreads, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
+        new ThreadFactoryBuilder().setNameFormat("FSTableDescriptorLoad-pool-%d").setDaemon(true)
+          .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
+      executor.allowCoreThreadTimeOut(true);
+    }
   }
 
   public static void tryUpdateMetaTableDescriptor(Configuration conf) throws IOException {
@@ -235,27 +258,56 @@ public class FSTableDescriptors implements TableDescriptors {
    */
   @Override
   public Map<String, TableDescriptor> getAll() throws IOException {
-    Map<String, TableDescriptor> tds = new TreeMap<>();
+    Map<String, TableDescriptor> tds = new ConcurrentSkipListMap<>();
     if (fsvisited) {
       for (Map.Entry<TableName, TableDescriptor> entry : this.cache.entrySet()) {
         tds.put(entry.getKey().getNameWithNamespaceInclAsString(), entry.getValue());
       }
     } else {
-      LOG.trace("Fetching table descriptors from the filesystem.");
-      boolean allvisited = usecache;
-      for (Path d : FSUtils.getTableDirs(fs, rootdir)) {
-        TableDescriptor htd = get(CommonFSUtils.getTableName(d));
-        if (htd == null) {
-          allvisited = false;
-        } else {
-          tds.put(htd.getTableName().getNameWithNamespaceInclAsString(), htd);
+      LOG.info("Fetching table descriptors from the filesystem.");
+      final long startTime = EnvironmentEdgeManager.currentTime();
+      AtomicBoolean allvisited = new AtomicBoolean(usecache);
+      List<Path> tableDirs = FSUtils.getTableDirs(fs, rootdir);
+      if (!tableDescriptorParallelLoadEnable) {
+        for (Path dir : tableDirs) {
+          internalGet(dir, tds, allvisited);
+        }
+      } else {
+        CountDownLatch latch = new CountDownLatch(tableDirs.size());
+        for (Path dir : tableDirs) {
+          executor.submit(new Runnable() {
+            @Override
+            public void run() {
+              try {
+                internalGet(dir, tds, allvisited);
+              } finally {
+                latch.countDown();
+              }
+            }
+          });
+        }
+        try {
+          latch.await();
+        } catch (InterruptedException ie) {
+          throw (InterruptedIOException) new InterruptedIOException().initCause(ie);
         }
       }
-      fsvisited = allvisited;
+      fsvisited = allvisited.get();
+      LOG.info("Fetched table descriptors(size=" + tds.size() + ") cost "
+        + (EnvironmentEdgeManager.currentTime() - startTime) + "ms.");
     }
     return tds;
   }
 
+  private void internalGet(Path dir, Map<String, TableDescriptor> tds, AtomicBoolean allvisited) {
+    TableDescriptor htd = get(CommonFSUtils.getTableName(dir));
+    if (htd == null) {
+      allvisited.set(false);
+    } else {
+      tds.put(htd.getTableName().getNameWithNamespaceInclAsString(), htd);
+    }
+  }
+
   /**
    * Find descriptors by namespace.
    * @see #get(org.apache.hadoop.hbase.TableName)
@@ -379,6 +431,14 @@ public class FSTableDescriptors implements TableDescriptors {
     return Bytes.toString(b);
   }
 
+  @Override
+  public void close() throws IOException {
+    // Close the executor when parallel loading enabled.
+    if (tableDescriptorParallelLoadEnable) {
+      this.executor.shutdown();
+    }
+  }
+
   static final class SequenceIdAndFileLength {
 
     final int sequenceId;

hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestFSTableDescriptors.java

@@ -285,6 +285,32 @@ public class TestFSTableDescriptors {
       + htds.getAll().size(), count + 1, htds.getAll().size());
   }
 
+  @Test
+  public void testParallelGetAll() throws IOException, InterruptedException {
+    final String name = "testParallelGetAll";
+    FileSystem fs = FileSystem.get(UTIL.getConfiguration());
+    // Enable parallel load table descriptor.
+    FSTableDescriptors htds = new FSTableDescriptorsTest(fs, testDir, true, 20);
+    final int count = 100;
+    // Write out table infos.
+    for (int i = 0; i < count; i++) {
+      htds.createTableDescriptor(
+        TableDescriptorBuilder.newBuilder(TableName.valueOf(name + i)).build());
+    }
+    // add hbase:meta
+    htds
+      .createTableDescriptor(TableDescriptorBuilder.newBuilder(TableName.META_TABLE_NAME).build());
+
+    int getTableDescriptorSize = htds.getAll().size();
+    assertEquals("getAll() didn't return all TableDescriptors, expected: " + (count + 1) + " got: "
+      + getTableDescriptorSize, count + 1, getTableDescriptorSize);
+
+    // get again to check whether the cache works well
+    getTableDescriptorSize = htds.getAll().size();
+    assertEquals("getAll() didn't return all TableDescriptors with cache, expected: " + (count + 1)
+      + " got: " + getTableDescriptorSize, count + 1, getTableDescriptorSize);
+  }
+
   @Test
   public void testGetAllOrdering() throws Exception {
     FileSystem fs = FileSystem.get(UTIL.getConfiguration());
@@ -467,6 +493,11 @@ public class TestFSTableDescriptors {
       super(fs, rootdir, false, usecache);
     }
 
+    public FSTableDescriptorsTest(FileSystem fs, Path rootdir, boolean usecache,
+      int tableDescriptorParallelLoadThreads) {
+      super(fs, rootdir, false, usecache, tableDescriptorParallelLoadThreads);
+    }
+
     @Override
     public TableDescriptor get(TableName tablename) {
       LOG.info((super.isUsecache() ? "Cached" : "Non-Cached") + " TableDescriptor.get() on "