Skip to content

Commit c2e01f2

Browse files
bharathvbinlijin
bharathv
authored and
binlijin
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>
1 parent 264d3e3 commit c2e01f2

File tree

3 files changed

+258
-0
lines changed

3 files changed

+258
-0
lines changed

‎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: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -451,6 +451,9 @@ public void run() {
451451
private final boolean maintenanceMode;
452452
static final String MAINTENANCE_MODE = "hbase.master.maintenance_mode";
453453

454+
// Cached clusterId on stand by masters to serve clusterID requests from clients.
455+
private final CachedClusterId cachedClusterId;
456+
454457
public static class RedirectServlet extends HttpServlet {
455458
private static final long serialVersionUID = 2894774810058302473L;
456459
private final int regionServerInfoPort;
@@ -571,6 +574,7 @@ public HMaster(final Configuration conf)
571574
} else {
572575
this.activeMasterManager = null;
573576
}
577+
cachedClusterId = new CachedClusterId(conf);
574578
} catch (Throwable t) {
575579
// Make sure we log the exception. HMaster is often started via reflection and the
576580
// cause of failed startup is lost.
@@ -3856,4 +3860,13 @@ public Map<String, ReplicationStatus> getWalGroupsReplicationStatus() {
38563860
public HbckChore getHbckChore() {
38573861
return this.hbckChore;
38583862
}
3863+
3864+
@Override
3865+
public String getClusterId() {
3866+
if (activeMaster) {
3867+
return super.getClusterId();
3868+
}
3869+
return cachedClusterId.getFromCacheOrFetch();
3870+
}
3871+
38593872
}

‎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)