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

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

@@ -444,6 +444,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;
@@ -564,6 +567,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.
@@ -3776,4 +3780,11 @@ public void runReplicationBarrierCleaner() {
   public SnapshotQuotaObserverChore getSnapshotQuotaObserverChore() {
     return this.snapshotQuotaChore;
   }
+
+  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);
+  }
+}