Skip to content

Commit 9ab6529

Browse files
bharathvbharathv
committed
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> (cherry picked from commit c2e01f2)
1 parent 9bee77b commit 9ab6529

3 files changed

Lines changed: 256 additions & 0 deletions

File tree

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

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.hadoop.hbase.master;
20+
21+
import java.io.IOException;
22+
import java.util.concurrent.atomic.AtomicBoolean;
23+
import java.util.concurrent.atomic.AtomicInteger;
24+
import org.apache.hadoop.conf.Configuration;
25+
import org.apache.hadoop.fs.FileSystem;
26+
import org.apache.hadoop.fs.Path;
27+
import org.apache.hadoop.hbase.ClusterId;
28+
import org.apache.hadoop.hbase.util.FSUtils;
29+
import org.apache.yetus.audience.InterfaceAudience;
30+
import org.slf4j.Logger;
31+
import org.slf4j.LoggerFactory;
32+
import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
33+
import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
34+
35+
/**
36+
* Caches the cluster ID of the cluster. For standby masters, this is used to serve the client
37+
* RPCs that fetch the cluster ID. ClusterID is only created by an active master if one does not
38+
* already exist. Standby masters just read the information from the file system. This class is
39+
* thread-safe.
40+
*
41+
* TODO: Make it a singleton without affecting concurrent junit tests.
42+
*/
43+
@InterfaceAudience.Private
44+
public class CachedClusterId {
45+
46+
public static final Logger LOG = LoggerFactory.getLogger(CachedClusterId.class);
47+
private static final int MAX_FETCH_TIMEOUT_MS = 10000;
48+
49+
private Path rootDir;
50+
private FileSystem fs;
51+
52+
// When true, indicates that a FileSystem fetch of ClusterID is in progress. This is used to
53+
// avoid multiple fetches from FS and let only one thread fetch the information.
54+
AtomicBoolean fetchInProgress = new AtomicBoolean(false);
55+
56+
// When true, it means that the cluster ID has been fetched successfully from fs.
57+
private AtomicBoolean isClusterIdSet = new AtomicBoolean(false);
58+
// Immutable once set and read multiple times.
59+
private ClusterId clusterId;
60+
61+
// cache stats for testing.
62+
private AtomicInteger cacheMisses = new AtomicInteger(0);
63+
64+
public CachedClusterId(Configuration conf) throws IOException {
65+
rootDir = FSUtils.getRootDir(conf);
66+
fs = rootDir.getFileSystem(conf);
67+
}
68+
69+
/**
70+
* Succeeds only once, when setting to a non-null value. Overwrites are not allowed.
71+
*/
72+
private void setClusterId(ClusterId id) {
73+
if (id == null || isClusterIdSet.get()) {
74+
return;
75+
}
76+
clusterId = id;
77+
isClusterIdSet.set(true);
78+
}
79+
80+
/**
81+
* Returns a cached copy of the cluster ID. null if the cache is not populated.
82+
*/
83+
private String getClusterId() {
84+
if (!isClusterIdSet.get()) {
85+
return null;
86+
}
87+
// It is ok to read without a lock since clusterId is immutable once set.
88+
return clusterId.toString();
89+
}
90+
91+
/**
92+
* Attempts to fetch the cluster ID from the file system. If no attempt is already in progress,
93+
* synchronously fetches the cluster ID and sets it. If an attempt is already in progress,
94+
* returns right away and the caller is expected to wait for the fetch to finish.
95+
* @return true if the attempt is done, false if another thread is already fetching it.
96+
*/
97+
private boolean attemptFetch() {
98+
if (fetchInProgress.compareAndSet(false, true)) {
99+
// A fetch is not in progress, so try fetching the cluster ID synchronously and then notify
100+
// the waiting threads.
101+
try {
102+
cacheMisses.incrementAndGet();
103+
setClusterId(FSUtils.getClusterId(fs, rootDir));
104+
} catch (IOException e) {
105+
LOG.warn("Error fetching cluster ID", e);
106+
} finally {
107+
Preconditions.checkState(fetchInProgress.compareAndSet(true, false));
108+
synchronized (fetchInProgress) {
109+
fetchInProgress.notifyAll();
110+
}
111+
}
112+
return true;
113+
}
114+
return false;
115+
}
116+
117+
private void waitForFetchToFinish() throws InterruptedException {
118+
synchronized (fetchInProgress) {
119+
while (fetchInProgress.get()) {
120+
// We don't want the fetches to block forever, for example if there are bugs
121+
// of missing notifications.
122+
fetchInProgress.wait(MAX_FETCH_TIMEOUT_MS);
123+
}
124+
}
125+
}
126+
127+
/**
128+
* Fetches the ClusterId from FS if it is not cached locally. Atomically updates the cached
129+
* copy and is thread-safe. Optimized to do a single fetch when there are multiple threads are
130+
* trying get from a clean cache.
131+
*
132+
* @return ClusterId by reading from FileSystem or null in any error case or cluster ID does
133+
* not exist on the file system.
134+
*/
135+
public String getFromCacheOrFetch() {
136+
String id = getClusterId();
137+
if (id != null) {
138+
return id;
139+
}
140+
if (!attemptFetch()) {
141+
// A fetch is in progress.
142+
try {
143+
waitForFetchToFinish();
144+
} catch (InterruptedException e) {
145+
// pass and return whatever is in the cache.
146+
}
147+
}
148+
return getClusterId();
149+
}
150+
151+
@VisibleForTesting
152+
public int getCacheStats() {
153+
return cacheMisses.get();
154+
}
155+
}

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -444,6 +444,9 @@ public void run() {
444444
private final boolean maintenanceMode;
445445
static final String MAINTENANCE_MODE = "hbase.master.maintenance_mode";
446446

447+
// Cached clusterId on stand by masters to serve clusterID requests from clients.
448+
private final CachedClusterId cachedClusterId;
449+
447450
public static class RedirectServlet extends HttpServlet {
448451
private static final long serialVersionUID = 2894774810058302473L;
449452
private final int regionServerInfoPort;
@@ -564,6 +567,7 @@ public HMaster(final Configuration conf)
564567
} else {
565568
this.activeMasterManager = null;
566569
}
570+
cachedClusterId = new CachedClusterId(conf);
567571
} catch (Throwable t) {
568572
// Make sure we log the exception. HMaster is often started via reflection and the
569573
// cause of failed startup is lost.
@@ -3791,4 +3795,11 @@ public void runReplicationBarrierCleaner() {
37913795
public SnapshotQuotaObserverChore getSnapshotQuotaObserverChore() {
37923796
return this.snapshotQuotaChore;
37933797
}
3798+
3799+
public String getClusterId() {
3800+
if (activeMaster) {
3801+
return super.getClusterId();
3802+
}
3803+
return cachedClusterId.getFromCacheOrFetch();
3804+
}
37943805
}

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

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.hadoop.hbase;
19+
20+
import static org.junit.Assert.assertEquals;
21+
import org.apache.hadoop.conf.Configuration;
22+
import org.apache.hadoop.hbase.MultithreadedTestUtil.TestContext;
23+
import org.apache.hadoop.hbase.MultithreadedTestUtil.TestThread;
24+
import org.apache.hadoop.hbase.master.CachedClusterId;
25+
import org.apache.hadoop.hbase.master.HMaster;
26+
import org.apache.hadoop.hbase.testclassification.MediumTests;
27+
import org.junit.AfterClass;
28+
import org.junit.BeforeClass;
29+
import org.junit.ClassRule;
30+
import org.junit.Test;
31+
import org.junit.experimental.categories.Category;
32+
33+
@Category(MediumTests.class)
34+
public class TestCachedClusterId {
35+
@ClassRule
36+
public static final HBaseClassTestRule CLASS_RULE =
37+
HBaseClassTestRule.forClass(TestCachedClusterId.class);
38+
39+
private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
40+
41+
private static String clusterId;
42+
private static HMaster activeMaster;
43+
private static HMaster standByMaster;
44+
45+
private static class GetClusterIdThread extends TestThread {
46+
CachedClusterId cachedClusterId;
47+
public GetClusterIdThread(TestContext ctx, CachedClusterId clusterId) {
48+
super(ctx);
49+
cachedClusterId = clusterId;
50+
}
51+
52+
@Override
53+
public void doWork() throws Exception {
54+
assertEquals(clusterId, cachedClusterId.getFromCacheOrFetch());
55+
}
56+
}
57+
58+
@BeforeClass
59+
public static void setUp() throws Exception {
60+
TEST_UTIL.startMiniCluster(1);
61+
activeMaster = TEST_UTIL.getHBaseCluster().getMaster();
62+
clusterId = activeMaster.getClusterId();
63+
standByMaster = TEST_UTIL.getHBaseCluster().startMaster().getMaster();
64+
}
65+
66+
@AfterClass
67+
public static void tearDown() throws Exception {
68+
TEST_UTIL.shutdownMiniCluster();
69+
}
70+
71+
@Test
72+
public void testClusterIdMatch() {
73+
assertEquals(clusterId, standByMaster.getClusterId());
74+
}
75+
76+
@Test
77+
public void testMultiThreadedGetClusterId() throws Exception {
78+
Configuration conf = TEST_UTIL.getConfiguration();
79+
CachedClusterId cachedClusterId = new CachedClusterId(conf);
80+
TestContext context = new TestContext(conf);
81+
int numThreads = 100;
82+
for (int i = 0; i < numThreads; i++) {
83+
context.addThread(new GetClusterIdThread(context, cachedClusterId));
84+
}
85+
context.startThreads();
86+
context.stop();
87+
int cacheMisses = cachedClusterId.getCacheStats();
88+
assertEquals(cacheMisses, 1);
89+
}
90+
}

0 commit comments

Comments
 (0)