HBASE-23257: Track clusterID in stand by masters (#798)

This patch implements a simple cache that all the masters
can lookup to serve cluster ID to clients. Active HMaster
is still responsible for creating it but all the masters
will read it from fs to serve clients.

RPCs exposing it will come in a separate patch as a part of
HBASE-18095.

Signed-off-by: Andrew Purtell <apurtell@apache.org>
Signed-off-by: Wellington Chevreuil <wchevreuil@apache.org>
Signed-off-by: Guangxu Cheng <guangxucheng@gmail.com>

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

@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.ClusterId;
+import org.apache.hadoop.hbase.util.FSUtils;
+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.hbase.thirdparty.com.google.common.base.Preconditions;
+
+/**
+ * Caches the cluster ID of the cluster. For standby masters, this is used to serve the client
+ * RPCs that fetch the cluster ID. ClusterID is only created by an active master if one does not
+ * already exist. Standby masters just read the information from the file system. This class is
+ * thread-safe.
+ *
+ * TODO: Make it a singleton without affecting concurrent junit tests.
+ */
+@InterfaceAudience.Private
+public class CachedClusterId {
+
+  public static final Logger LOG = LoggerFactory.getLogger(CachedClusterId.class);
+  private static final int MAX_FETCH_TIMEOUT_MS = 10000;
+
+  private Path rootDir;
+  private FileSystem fs;
+
+  // When true, indicates that a FileSystem fetch of ClusterID is in progress. This is used to
+  // avoid multiple fetches from FS and let only one thread fetch the information.
+  AtomicBoolean fetchInProgress = new AtomicBoolean(false);
+
+  // When true, it means that the cluster ID has been fetched successfully from fs.
+  private AtomicBoolean isClusterIdSet = new AtomicBoolean(false);
+  // Immutable once set and read multiple times.
+  private ClusterId clusterId;
+
+  // cache stats for testing.
+  private AtomicInteger cacheMisses = new AtomicInteger(0);
+
+  public CachedClusterId(Configuration conf) throws IOException {
+    rootDir = FSUtils.getRootDir(conf);
+    fs = rootDir.getFileSystem(conf);
+  }
+
+  /**
+   * Succeeds only once, when setting to a non-null value. Overwrites are not allowed.
+   */
+  private void setClusterId(ClusterId id) {
+    if (id == null || isClusterIdSet.get()) {
+      return;
+    }
+    clusterId = id;
+    isClusterIdSet.set(true);
+  }
+
+  /**
+   * Returns a cached copy of the cluster ID. null if the cache is not populated.
+   */
+  private String getClusterId() {
+    if (!isClusterIdSet.get()) {
+      return null;
+    }
+    // It is ok to read without a lock since clusterId is immutable once set.
+    return clusterId.toString();
+  }
+
+  /**
+   * Attempts to fetch the cluster ID from the file system. If no attempt is already in progress,
+   * synchronously fetches the cluster ID and sets it. If an attempt is already in progress,
+   * returns right away and the caller is expected to wait for the fetch to finish.
+   * @return true if the attempt is done, false if another thread is already fetching it.
+   */
+  private boolean attemptFetch() {
+    if (fetchInProgress.compareAndSet(false, true)) {
+      // A fetch is not in progress, so try fetching the cluster ID synchronously and then notify
+      // the waiting threads.
+      try {
+        cacheMisses.incrementAndGet();
+        setClusterId(FSUtils.getClusterId(fs, rootDir));
+      } catch (IOException e) {
+        LOG.warn("Error fetching cluster ID", e);
+      } finally {
+        Preconditions.checkState(fetchInProgress.compareAndSet(true, false));
+        synchronized (fetchInProgress) {
+          fetchInProgress.notifyAll();
+        }
+      }
+      return true;
+    }
+    return false;
+  }
+
+  private void waitForFetchToFinish() throws InterruptedException {
+    synchronized (fetchInProgress) {
+      while (fetchInProgress.get()) {
+        // We don't want the fetches to block forever, for example if there are bugs
+        // of missing notifications.
+        fetchInProgress.wait(MAX_FETCH_TIMEOUT_MS);
+      }
+    }
+  }
+
+  /**
+   * Fetches the ClusterId from FS if it is not cached locally. Atomically updates the cached
+   * copy and is thread-safe. Optimized to do a single fetch when there are multiple threads are
+   * trying get from a clean cache.
+   *
+   * @return ClusterId by reading from FileSystem or null in any error case or cluster ID does
+   *     not exist on the file system.
+   */
+  public String getFromCacheOrFetch() {
+    String id = getClusterId();
+    if (id != null) {
+      return id;
+    }
+    if (!attemptFetch()) {
+      // A fetch is in progress.
+      try {
+        waitForFetchToFinish();
+      } catch (InterruptedException e) {
+        // pass and return whatever is in the cache.
+      }
+    }
+    return getClusterId();
+  }
+
+  @VisibleForTesting
+  public int getCacheStats() {
+    return cacheMisses.get();
+  }
+}

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

@@ -451,6 +451,9 @@ public void run() {
   private final boolean maintenanceMode;
   static final String MAINTENANCE_MODE = "hbase.master.maintenance_mode";
 
+  // Cached clusterId on stand by masters to serve clusterID requests from clients.
+  private final CachedClusterId cachedClusterId;
+
   public static class RedirectServlet extends HttpServlet {
     private static final long serialVersionUID = 2894774810058302473L;
     private final int regionServerInfoPort;
@@ -571,6 +574,7 @@ public HMaster(final Configuration conf)
       } else {
         this.activeMasterManager = null;
       }
+      cachedClusterId = new CachedClusterId(conf);
     } catch (Throwable t) {
       // Make sure we log the exception. HMaster is often started via reflection and the
       // cause of failed startup is lost.
@@ -3856,4 +3860,13 @@ public Map<String, ReplicationStatus> getWalGroupsReplicationStatus() {
   public HbckChore getHbckChore() {
     return this.hbckChore;
   }
+
+  @Override
+  public String getClusterId() {
+    if (activeMaster) {
+      return super.getClusterId();
+    }
+    return cachedClusterId.getFromCacheOrFetch();
+  }
+
 }

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

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase;
+
+import static org.junit.Assert.assertEquals;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.MultithreadedTestUtil.TestContext;
+import org.apache.hadoop.hbase.MultithreadedTestUtil.TestThread;
+import org.apache.hadoop.hbase.master.CachedClusterId;
+import org.apache.hadoop.hbase.master.HMaster;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(MediumTests.class)
+public class TestCachedClusterId {
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+      HBaseClassTestRule.forClass(TestCachedClusterId.class);
+
+  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+
+  private static String clusterId;
+  private static HMaster activeMaster;
+  private static HMaster standByMaster;
+
+  private static class GetClusterIdThread extends TestThread {
+    CachedClusterId cachedClusterId;
+    public GetClusterIdThread(TestContext ctx, CachedClusterId clusterId) {
+      super(ctx);
+      cachedClusterId = clusterId;
+    }
+
+    @Override
+    public void doWork() throws Exception {
+      assertEquals(clusterId, cachedClusterId.getFromCacheOrFetch());
+    }
+  }
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    TEST_UTIL.startMiniCluster(1);
+    activeMaster = TEST_UTIL.getHBaseCluster().getMaster();
+    clusterId = activeMaster.getClusterId();
+    standByMaster = TEST_UTIL.getHBaseCluster().startMaster().getMaster();
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Test
+  public void testClusterIdMatch() {
+    assertEquals(clusterId, standByMaster.getClusterId());
+  }
+
+  @Test
+  public void testMultiThreadedGetClusterId() throws Exception {
+    Configuration conf = TEST_UTIL.getConfiguration();
+    CachedClusterId cachedClusterId = new CachedClusterId(conf);
+    TestContext context = new TestContext(conf);
+    int numThreads = 100;
+    for (int i = 0; i < numThreads; i++) {
+      context.addThread(new GetClusterIdThread(context, cachedClusterId));
+    }
+    context.startThreads();
+    context.stop();
+    int cacheMisses = cachedClusterId.getCacheStats();
+    assertEquals(cacheMisses, 1);
+  }
+}