Skip to content

Commit 2e94f6f

Browse files
vaijoshvaijosh
authored
HBASE-27527 Port HBASE-27498 to branch-2 (#4919)
 Signed-off-by: Tak Lon (Stephen) Wu <taklwu@apache.org>
1 parent 98d6682 commit 2e94f6f

File tree

2 files changed

+221
-21
lines changed

2 files changed

+221
-21
lines changed

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java‎

Lines changed: 61 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@
9696
import org.slf4j.Logger;
9797
import org.slf4j.LoggerFactory;
9898

99+
import org.apache.hbase.thirdparty.com.google.common.base.Suppliers;
99100
import org.apache.hbase.thirdparty.com.google.common.base.Throwables;
100101
import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
101102
import org.apache.hbase.thirdparty.com.google.protobuf.BlockingRpcChannel;
@@ -179,6 +180,9 @@
179180
@InterfaceAudience.Private
180181
public class ConnectionImplementation implements ClusterConnection, Closeable {
181182
public static final String RETRIES_BY_SERVER_KEY = "hbase.client.retries.by.server";
183+
184+
public static final String MASTER_STATE_CACHE_TIMEOUT_SEC =
185+
"hbase.client.master.state.cache.timeout.sec";
182186
private static final Logger LOG = LoggerFactory.getLogger(ConnectionImplementation.class);
183187

184188
// The mode tells if HedgedRead, LoadBalance mode is supported.
@@ -261,6 +265,12 @@ public class ConnectionImplementation implements ClusterConnection, Closeable {
261265
/** lock guards against multiple threads trying to query the meta region at the same time */
262266
private final ReentrantLock userRegionLock = new ReentrantLock();
263267

268+
/**
269+
* Supplier to get masterState.By default uses simple supplier without TTL cache. When
270+
* hbase.client.master.state.cache.timeout.sec > 0 it uses TTL Cache.
271+
*/
272+
private final Supplier<Boolean> masterStateSupplier;
273+
264274
private ChoreService choreService;
265275

266276
/**
@@ -395,6 +405,39 @@ public void newDead(ServerName sn) {
395405
default:
396406
// Doing nothing
397407
}
408+
409+
long masterStateCacheTimeout = conf.getLong(MASTER_STATE_CACHE_TIMEOUT_SEC, 0);
410+
411+
Supplier<Boolean> masterConnSupplier = masterConnectionStateSupplier();
412+
if (masterStateCacheTimeout <= 0L) {
413+
this.masterStateSupplier = masterConnSupplier;
414+
} else {
415+
this.masterStateSupplier = Suppliers.memoizeWithExpiration(masterConnSupplier::get,
416+
masterStateCacheTimeout, TimeUnit.SECONDS);
417+
}
418+
}
419+
420+
/**
421+
* Visible for tests
422+
*/
423+
Supplier<Boolean> masterConnectionStateSupplier() {
424+
return () -> {
425+
if (this.masterServiceState.getStub() == null) {
426+
return false;
427+
}
428+
try {
429+
LOG.info("Getting master state using rpc call");
430+
return this.masterServiceState.isMasterRunning();
431+
} catch (UndeclaredThrowableException e) {
432+
// It's somehow messy, but we can receive exceptions such as
433+
// java.net.ConnectException but they're not declared. So we catch it...
434+
LOG.info("Master connection is not running anymore", e.getUndeclaredThrowable());
435+
return false;
436+
} catch (IOException se) {
437+
LOG.warn("Checking master connection", se);
438+
return false;
439+
}
440+
};
398441
}
399442

400443
private void spawnRenewalChore(final UserGroupInformation user) {
@@ -1268,7 +1311,6 @@ public void addError() {
12681311
* Class to make a MasterServiceStubMaker stub.
12691312
*/
12701313
private final class MasterServiceStubMaker {
1271-
12721314
private void isMasterRunning(MasterProtos.MasterService.BlockingInterface stub)
12731315
throws IOException {
12741316
try {
@@ -1368,6 +1410,13 @@ public BlockingInterface getClient(ServerName serverName) throws IOException {
13681410

13691411
final MasterServiceState masterServiceState = new MasterServiceState(this);
13701412

1413+
/**
1414+
* Visible for tests
1415+
*/
1416+
MasterServiceState getMasterServiceState() {
1417+
return this.masterServiceState;
1418+
}
1419+
13711420
@Override
13721421
public MasterKeepAliveConnection getMaster() throws IOException {
13731422
return getKeepAliveMasterService();
@@ -1378,13 +1427,16 @@ private void resetMasterServiceState(final MasterServiceState mss) {
13781427
}
13791428

13801429
private MasterKeepAliveConnection getKeepAliveMasterService() throws IOException {
1381-
synchronized (masterLock) {
1382-
if (!isKeepAliveMasterConnectedAndRunning(this.masterServiceState)) {
1383-
MasterServiceStubMaker stubMaker = new MasterServiceStubMaker();
1384-
this.masterServiceState.stub = stubMaker.makeStub();
1430+
if (!isKeepAliveMasterConnectedAndRunning()) {
1431+
synchronized (masterLock) {
1432+
if (!isKeepAliveMasterConnectedAndRunning()) {
1433+
MasterServiceStubMaker stubMaker = new MasterServiceStubMaker();
1434+
this.masterServiceState.stub = stubMaker.makeStub();
1435+
}
1436+
resetMasterServiceState(this.masterServiceState);
13851437
}
1386-
resetMasterServiceState(this.masterServiceState);
13871438
}
1439+
13881440
// Ugly delegation just so we can add in a Close method.
13891441
final MasterProtos.MasterService.BlockingInterface stub = this.masterServiceState.stub;
13901442
return new MasterKeepAliveConnection() {
@@ -1977,21 +2029,9 @@ private static void release(MasterServiceState mss) {
19772029
}
19782030
}
19792031

1980-
private boolean isKeepAliveMasterConnectedAndRunning(MasterServiceState mss) {
1981-
if (mss.getStub() == null) {
1982-
return false;
1983-
}
1984-
try {
1985-
return mss.isMasterRunning();
1986-
} catch (UndeclaredThrowableException e) {
1987-
// It's somehow messy, but we can receive exceptions such as
1988-
// java.net.ConnectException but they're not declared. So we catch it...
1989-
LOG.info("Master connection is not running anymore", e.getUndeclaredThrowable());
1990-
return false;
1991-
} catch (IOException se) {
1992-
LOG.warn("Checking master connection", se);
1993-
return false;
1994-
}
2032+
private boolean isKeepAliveMasterConnectedAndRunning() {
2033+
LOG.info("Getting master connection state from TTL Cache");
2034+
return masterStateSupplier.get();
19952035
}
19962036

19972037
void releaseMaster(MasterServiceState mss) {

‎hbase-it/src/test/java/org/apache/hadoop/hbase/client/TestConnectionImplementation.java‎

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
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.client;
19+
20+
import java.io.IOException;
21+
import java.lang.reflect.InvocationTargetException;
22+
import java.lang.reflect.Method;
23+
import java.lang.reflect.UndeclaredThrowableException;
24+
import org.apache.commons.lang3.reflect.FieldUtils;
25+
import org.apache.hadoop.conf.Configuration;
26+
import org.apache.hadoop.hbase.HBaseClassTestRule;
27+
import org.apache.hadoop.hbase.IntegrationTestingUtility;
28+
import org.apache.hadoop.hbase.security.UserProvider;
29+
import org.apache.hadoop.hbase.testclassification.ClientTests;
30+
import org.apache.hadoop.hbase.testclassification.MediumTests;
31+
import org.junit.AfterClass;
32+
import org.junit.Assert;
33+
import org.junit.BeforeClass;
34+
import org.junit.ClassRule;
35+
import org.junit.Test;
36+
import org.junit.experimental.categories.Category;
37+
import org.junit.runner.RunWith;
38+
import org.mockito.Mockito;
39+
import org.mockito.junit.MockitoJUnitRunner;
40+
41+
@Category({ ClientTests.class, MediumTests.class })
42+
@RunWith(MockitoJUnitRunner.class)
43+
public class TestConnectionImplementation {
44+
@ClassRule
45+
public static final HBaseClassTestRule CLASS_RULE =
46+
HBaseClassTestRule.forClass(TestConnectionImplementation.class);
47+
private static final IntegrationTestingUtility TEST_UTIL = new IntegrationTestingUtility();
48+
49+
@BeforeClass
50+
public static void beforeClass() throws Exception {
51+
TEST_UTIL.startMiniCluster(1);
52+
}
53+
54+
@AfterClass
55+
public static void afterClass() throws Exception {
56+
TEST_UTIL.shutdownMiniCluster();
57+
}
58+
59+
@Test
60+
public void testGetMaster_noCachedMasterState() throws IOException, IllegalAccessException {
61+
Configuration conf = TEST_UTIL.getConfiguration();
62+
conf.setLong(ConnectionImplementation.MASTER_STATE_CACHE_TIMEOUT_SEC, 0L);
63+
ConnectionImplementation conn =
64+
new ConnectionImplementation(conf, null, UserProvider.instantiate(conf).getCurrent());
65+
ConnectionImplementation.MasterServiceState masterServiceState = spyMasterServiceState(conn);
66+
conn.getMaster(); // This initializes the stubs but don't call isMasterRunning
67+
conn.getMaster(); // Calls isMasterRunning since stubs are initialized. Invocation 1
68+
conn.getMaster(); // Calls isMasterRunning since stubs are initialized. Invocation 2
69+
Mockito.verify(masterServiceState, Mockito.times(2)).isMasterRunning();
70+
conn.close();
71+
}
72+
73+
@Test
74+
public void testGetMaster_masterStateCacheHit() throws IOException, IllegalAccessException {
75+
Configuration conf = TEST_UTIL.getConfiguration();
76+
conf.setLong(ConnectionImplementation.MASTER_STATE_CACHE_TIMEOUT_SEC, 15L);
77+
ConnectionImplementation conn =
78+
new ConnectionImplementation(conf, null, UserProvider.instantiate(conf).getCurrent());
79+
ConnectionImplementation.MasterServiceState masterServiceState = spyMasterServiceState(conn);
80+
conn.getMaster(); // This initializes the stubs but don't call isMasterRunning
81+
conn.getMaster(); // Uses cached value, don't call isMasterRunning
82+
conn.getMaster(); // Uses cached value, don't call isMasterRunning
83+
Mockito.verify(masterServiceState, Mockito.times(0)).isMasterRunning();
84+
conn.close();
85+
}
86+
87+
@Test
88+
public void testGetMaster_masterStateCacheMiss()
89+
throws IOException, InterruptedException, IllegalAccessException {
90+
Configuration conf = TEST_UTIL.getConfiguration();
91+
conf.setLong(ConnectionImplementation.MASTER_STATE_CACHE_TIMEOUT_SEC, 5L);
92+
ConnectionImplementation conn =
93+
new ConnectionImplementation(conf, null, UserProvider.instantiate(conf).getCurrent());
94+
ConnectionImplementation.MasterServiceState masterServiceState = spyMasterServiceState(conn);
95+
conn.getMaster(); // This initializes the stubs but don't call isMasterRunning
96+
conn.getMaster(); // Uses cached value, don't call isMasterRunning
97+
conn.getMaster(); // Uses cached value, don't call isMasterRunning
98+
Thread.sleep(10000);
99+
conn.getMaster(); // Calls isMasterRunning after cache expiry. Invocation 1
100+
Mockito.verify(masterServiceState, Mockito.times(1)).isMasterRunning();
101+
conn.close();
102+
}
103+
104+
@Test
105+
public void testIsKeepAliveMasterConnectedAndRunning_UndeclaredThrowableException()
106+
throws IOException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
107+
Configuration conf = TEST_UTIL.getConfiguration();
108+
conf.setLong(ConnectionImplementation.MASTER_STATE_CACHE_TIMEOUT_SEC, 0);
109+
ConnectionImplementation conn =
110+
new ConnectionImplementation(conf, null, UserProvider.instantiate(conf).getCurrent());
111+
conn.getMaster(); // Initializes stubs
112+
113+
ConnectionImplementation.MasterServiceState masterServiceState = spyMasterServiceState(conn);
114+
Mockito.doThrow(new UndeclaredThrowableException(new Exception("DUMMY EXCEPTION")))
115+
.when(masterServiceState).isMasterRunning();
116+
117+
// Verify that masterState is "false" because of to injected exception
118+
boolean isKeepAliveMasterRunning =
119+
(boolean) getIsKeepAliveMasterConnectedAndRunningMethod().invoke(conn);
120+
Assert.assertFalse(isKeepAliveMasterRunning);
121+
conn.close();
122+
}
123+
124+
@Test
125+
public void testIsKeepAliveMasterConnectedAndRunning_IOException()
126+
throws IOException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
127+
Configuration conf = TEST_UTIL.getConfiguration();
128+
conf.setLong(ConnectionImplementation.MASTER_STATE_CACHE_TIMEOUT_SEC, 0);
129+
ConnectionImplementation conn =
130+
new ConnectionImplementation(conf, null, UserProvider.instantiate(conf).getCurrent());
131+
conn.getMaster();
132+
133+
ConnectionImplementation.MasterServiceState masterServiceState = spyMasterServiceState(conn);
134+
Mockito.doThrow(new IOException("DUMMY EXCEPTION")).when(masterServiceState).isMasterRunning();
135+
136+
boolean isKeepAliveMasterRunning =
137+
(boolean) getIsKeepAliveMasterConnectedAndRunningMethod().invoke(conn);
138+
139+
// Verify that masterState is "false" because of to injected exception
140+
Assert.assertFalse(isKeepAliveMasterRunning);
141+
conn.close();
142+
}
143+
144+
// Spy the masterServiceState object using reflection
145+
private ConnectionImplementation.MasterServiceState
146+
spyMasterServiceState(ConnectionImplementation conn) throws IllegalAccessException {
147+
ConnectionImplementation.MasterServiceState spiedMasterServiceState =
148+
Mockito.spy(conn.getMasterServiceState());
149+
FieldUtils.writeDeclaredField(conn, "masterServiceState", spiedMasterServiceState, true);
150+
return spiedMasterServiceState;
151+
}
152+
153+
// Get isKeepAliveMasterConnectedAndRunning using reflection
154+
private Method getIsKeepAliveMasterConnectedAndRunningMethod() throws NoSuchMethodException {
155+
Method method =
156+
ConnectionImplementation.class.getDeclaredMethod("isKeepAliveMasterConnectedAndRunning");
157+
method.setAccessible(true);
158+
return method;
159+
}
160+
}

0 commit comments

Comments
 (0)