Skip to content

Commit 98d80fc

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) (cherry picked from commit 9ab6529)
1 parent 20d0a32 commit 98d80fc

3 files changed

Lines changed: 252 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 com.google.common.annotations.VisibleForTesting;
22+
import com.google.common.base.Preconditions;
23+
import java.io.IOException;
24+
import java.util.concurrent.atomic.AtomicBoolean;
25+
import java.util.concurrent.atomic.AtomicInteger;
26+
import org.apache.hadoop.conf.Configuration;
27+
import org.apache.hadoop.fs.FileSystem;
28+
import org.apache.hadoop.fs.Path;
29+
import org.apache.hadoop.hbase.ClusterId;
30+
import org.apache.hadoop.hbase.classification.InterfaceAudience;
31+
import org.apache.hadoop.hbase.util.FSUtils;
32+
import org.slf4j.Logger;
33+
import org.slf4j.LoggerFactory;
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
@@ -386,6 +386,9 @@ public void run() {
386386
/** jetty server for master to redirect requests to regionserver infoServer */
387387
private org.mortbay.jetty.Server masterJettyServer;
388388

389+
// Cached clusterId on stand by masters to serve clusterID requests from clients.
390+
private final CachedClusterId cachedClusterId;
391+
389392
public static class RedirectServlet extends HttpServlet {
390393
private static final long serialVersionUID = 2894774810058302473L;
391394
private final int regionServerInfoPort;
@@ -521,6 +524,7 @@ public HMaster(final Configuration conf, CoordinatedStateManager csm)
521524
} else {
522525
activeMasterManager = null;
523526
}
527+
cachedClusterId = new CachedClusterId(conf);
524528
}
525529

526530
// return the actual infoPort, -1 means disable info server.
@@ -3429,4 +3433,11 @@ public LoadBalancer getLoadBalancer() {
34293433
}
34303434
return replicationLoadSourceMap;
34313435
}
3436+
3437+
public String getClusterId() {
3438+
if (activeMaster) {
3439+
return super.getClusterId();
3440+
}
3441+
return cachedClusterId.getFromCacheOrFetch();
3442+
}
34323443
}

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

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
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.Test;
30+
import org.junit.experimental.categories.Category;
31+
32+
@Category(MediumTests.class)
33+
public class TestCachedClusterId {
34+
35+
private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
36+
37+
private static String clusterId;
38+
private static HMaster activeMaster;
39+
private static HMaster standByMaster;
40+
41+
private static class GetClusterIdThread extends TestThread {
42+
CachedClusterId cachedClusterId;
43+
public GetClusterIdThread(TestContext ctx, CachedClusterId clusterId) {
44+
super(ctx);
45+
cachedClusterId = clusterId;
46+
}
47+
48+
@Override
49+
public void doWork() throws Exception {
50+
assertEquals(clusterId, cachedClusterId.getFromCacheOrFetch());
51+
}
52+
}
53+
54+
@BeforeClass
55+
public static void setUp() throws Exception {
56+
TEST_UTIL.startMiniCluster(1);
57+
activeMaster = TEST_UTIL.getHBaseCluster().getMaster();
58+
clusterId = activeMaster.getClusterId();
59+
standByMaster = TEST_UTIL.getHBaseCluster().startMaster().getMaster();
60+
}
61+
62+
@AfterClass
63+
public static void tearDown() throws Exception {
64+
TEST_UTIL.shutdownMiniCluster();
65+
}
66+
67+
@Test
68+
public void testClusterIdMatch() {
69+
assertEquals(clusterId, standByMaster.getClusterId());
70+
}
71+
72+
@Test
73+
public void testMultiThreadedGetClusterId() throws Exception {
74+
Configuration conf = TEST_UTIL.getConfiguration();
75+
CachedClusterId cachedClusterId = new CachedClusterId(conf);
76+
TestContext context = new TestContext(conf);
77+
int numThreads = 100;
78+
for (int i = 0; i < numThreads; i++) {
79+
context.addThread(new GetClusterIdThread(context, cachedClusterId));
80+
}
81+
context.startThreads();
82+
context.stop();
83+
int cacheMisses = cachedClusterId.getCacheStats();
84+
assertEquals(cacheMisses, 1);
85+
}
86+
}

0 commit comments

Comments
 (0)