Skip to content

Commit 9f03d15

Browse files
dengzhhu653dengzhhu653
committed
remove code smell
1 parent 9b583b3 commit 9f03d15

File tree

8 files changed

+266
-240
lines changed

8 files changed

+266
-240
lines changed

‎itests/hive-unit/src/test/java/org/apache/hadoop/hive/metastore/MetastoreHousekeepingLeaderTestBase.java‎

Lines changed: 131 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,15 @@
1818

1919
package org.apache.hadoop.hive.metastore;
2020

21+
import org.apache.commons.lang3.tuple.Pair;
2122
import org.apache.hadoop.conf.Configuration;
2223
import org.apache.hadoop.hdfs.MiniDFSCluster;
24+
import org.apache.hadoop.hive.common.TableName;
2325
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
2426
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
2527
import org.apache.hadoop.hive.metastore.leader.LeaderElection;
28+
import org.apache.hadoop.hive.metastore.leader.LeaderElectionContext;
29+
import org.apache.hadoop.hive.metastore.leader.LeaseLeaderElection;
2630
import org.apache.hadoop.hive.metastore.security.HadoopThriftAuthBridge;
2731
import org.apache.hadoop.hive.ql.stats.StatsUpdaterThread;
2832
import org.apache.hadoop.hive.ql.txn.compactor.Cleaner;
@@ -32,7 +36,10 @@
3236
import org.slf4j.Logger;
3337
import org.slf4j.LoggerFactory;
3438

39+
import java.io.IOException;
40+
import java.util.ArrayList;
3541
import java.util.HashMap;
42+
import java.util.List;
3643
import java.util.Map;
3744
import java.util.Set;
3845
import java.util.concurrent.CountDownLatch;
@@ -41,7 +48,7 @@
4148
/**
4249
* Base class for HMS leader config testing.
4350
*/
44-
public abstract class MetastoreHousekeepingLeaderTestBase {
51+
abstract class MetastoreHousekeepingLeaderTestBase {
4552
private static final Logger LOG = LoggerFactory.getLogger(MetastoreHousekeepingLeaderTestBase.class);
4653
private static HiveMetaStoreClient client;
4754
protected Configuration conf;
@@ -72,8 +79,6 @@ void setup(final String leaderHostName, Configuration configuration) throws Exce
7279

7380
warehouse = new Warehouse(conf);
7481

75-
conf.set("metastore.leader.test.listener", TestLeaderNotification.class.getName());
76-
7782
if (isServerStarted) {
7883
Assert.assertNotNull("Unable to connect to the MetaStore server", client);
7984
return;
@@ -201,32 +206,145 @@ private void resetThreadStatus() {
201206
threadClasses.forEach((thread, status) -> threadClasses.put(thread, false));
202207
}
203208

204-
public static class TestLeaderNotification implements LeaderElection.LeadershipStateListener {
209+
static class CombinedLeaderElector implements AutoCloseable {
210+
List<Pair<TableName, LeaderElection<TableName>>> elections = new ArrayList<>();
211+
private final Configuration configuration;
212+
private String name;
213+
214+
CombinedLeaderElector(Configuration conf) throws IOException {
215+
this.configuration = conf;
216+
for (LeaderElectionContext.TTYPE type : LeaderElectionContext.TTYPE.values()) {
217+
TableName table = type.getTableName();
218+
elections.add(Pair.of(table, new LeaseLeaderElection()));
219+
}
220+
}
221+
222+
public void tryBeLeader() throws Exception {
223+
int i = 0;
224+
for (Pair<TableName, LeaderElection<TableName>> election : elections) {
225+
LeaderElection<TableName> le = election.getRight();
226+
le.setName(name + "-" + i++);
227+
le.tryBeLeader(configuration, election.getLeft());
228+
}
229+
}
230+
231+
public boolean isLeader() {
232+
boolean isLeader = true;
233+
for (Pair<TableName, LeaderElection<TableName>> election : elections) {
234+
isLeader &= election.getRight().isLeader();
235+
}
236+
return isLeader;
237+
}
238+
239+
public void setName(String name) {
240+
this.name = name;
241+
}
242+
243+
@Override
244+
public void close() throws Exception {
245+
for (Pair<TableName, LeaderElection<TableName>> election : elections) {
246+
election.getRight().close();
247+
}
248+
}
249+
}
250+
251+
static class ReleaseAndRequireLease extends LeaseLeaderElection {
205252
private static CountDownLatch latch;
253+
private final Configuration configuration;
254+
private final boolean needRenewLease;
255+
private TableName tableName;
206256

207257
public static void setMonitor(CountDownLatch latch) {
208-
TestLeaderNotification.latch = latch;
258+
ReleaseAndRequireLease.latch = latch;
209259
}
210260
public static void reset() {
211-
TestLeaderNotification.latch = null;
261+
ReleaseAndRequireLease.latch = null;
262+
}
263+
264+
public ReleaseAndRequireLease(Configuration conf, boolean needRenewLease) throws IOException {
265+
super();
266+
this.configuration = conf;
267+
this.needRenewLease = needRenewLease;
212268
}
213269

214270
@Override
215-
public void takeLeadership(LeaderElection election) throws Exception {
216-
if (latch != null) {
217-
latch.countDown();
271+
public void setName(String name) {
272+
super.setName(name);
273+
LeaderElectionContext.TTYPE type = null;
274+
for (LeaderElectionContext.TTYPE value : LeaderElectionContext.TTYPE.values()) {
275+
if (value.getName().equalsIgnoreCase(name)) {
276+
type = value;
277+
break;
278+
}
279+
}
280+
if (type == null) {
281+
// This shouldn't happen at all
282+
throw new AssertionError("Unknown elector name: " + name);
218283
}
284+
this.tableName = type.getTableName();
219285
}
220286

221287
@Override
222-
public void lossLeadership(LeaderElection election) throws Exception {
288+
protected void notifyListener() {
289+
super.notifyListener();
290+
if (isLeader) {
291+
if (!needRenewLease) {
292+
super.shutdownWatcher();
293+
// In our tests, the time spent on notifying the listener might be greater than the lease timeout,
294+
// which makes the leader loss the leadership quickly after wake up, and kill all housekeeping services.
295+
// Make sure the leader is still valid while notifying the listener, and switch to ReleaseAndRequireWatcher
296+
// after all listeners finish their work.
297+
heartbeater = new ReleaseAndRequireWatcher(configuration, tableName);
298+
heartbeater.startWatch();
299+
}
300+
} else {
301+
try {
302+
// This is the last one get notified, sleep some time to make sure all other
303+
// services have been stopped before return
304+
Thread.sleep(12000);
305+
} catch (InterruptedException ignore) {
306+
}
307+
}
223308
if (latch != null) {
224-
// This is the last one get notified, sleep some time to make sure all other
225-
// services have been stopped before return
226-
Thread.sleep(12000);
227309
latch.countDown();
228310
}
229311
}
312+
313+
// For testing purpose only, lock would become timeout and then acquire it again
314+
private class ReleaseAndRequireWatcher extends LeaseWatcher {
315+
long timeout;
316+
public ReleaseAndRequireWatcher(Configuration conf,
317+
TableName tableName) {
318+
super(conf, tableName);
319+
timeout = MetastoreConf.getTimeVar(conf,
320+
MetastoreConf.ConfVars.TXN_TIMEOUT, TimeUnit.MILLISECONDS) + 3000;
321+
setName("ReleaseAndRequireWatcher-" + ((name != null) ? name + "-" : "") + ID.incrementAndGet());
322+
}
323+
324+
@Override
325+
public void beforeRun() {
326+
try {
327+
Thread.sleep(timeout);
328+
} catch (InterruptedException e) {
329+
// ignore this
330+
}
331+
}
332+
333+
@Override
334+
public void runInternal() {
335+
shutDown();
336+
// The timeout lock should be cleaned,
337+
// sleep some time to let others take the chance to become the leader
338+
try {
339+
Thread.sleep(5000);
340+
} catch (InterruptedException e) {
341+
// ignore
342+
}
343+
// Acquire the lock again
344+
conf = new Configuration(conf);
345+
reclaim();
346+
}
347+
}
230348
}
231349

232350
public void checkHouseKeepingThreadExistence(boolean isLeader) throws Exception {

‎itests/hive-unit/src/test/java/org/apache/hadoop/hive/metastore/TestMetastoreLeaseLeader.java‎

Lines changed: 20 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,9 @@
1919
package org.apache.hadoop.hive.metastore;
2020

2121
import org.apache.hadoop.conf.Configuration;
22-
import org.apache.hadoop.hive.common.TableName;
2322
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
24-
import org.apache.hadoop.hive.metastore.leader.LeaderElection;
2523
import org.apache.hadoop.hive.metastore.leader.LeaderElectionContext;
26-
import org.apache.hadoop.hive.metastore.leader.LeaseLeaderElection;
24+
import org.apache.hadoop.hive.metastore.leader.LeaderElectionFactory;
2725
import org.junit.After;
2826
import org.junit.Before;
2927
import org.junit.Test;
@@ -36,52 +34,50 @@
3634

3735
public class TestMetastoreLeaseLeader extends MetastoreHousekeepingLeaderTestBase {
3836

39-
LeaderElection<TableName> election;
37+
CombinedLeaderElector elector;
4038

4139
@Before
4240
public void setUp() throws Exception {
4341
Configuration configuration = MetastoreConf.newMetastoreConf();
44-
MetastoreConf.setTimeVar(configuration, MetastoreConf.ConfVars.TXN_TIMEOUT, 1, TimeUnit.SECONDS);
42+
MetastoreConf.setTimeVar(configuration, MetastoreConf.ConfVars.TXN_TIMEOUT, 3, TimeUnit.SECONDS);
4543
MetastoreConf.setTimeVar(configuration, MetastoreConf.ConfVars.LOCK_SLEEP_BETWEEN_RETRIES, 200, TimeUnit.MILLISECONDS);
46-
configuration.setBoolean(LeaseLeaderElection.METASTORE_RENEW_LEASE, false);
47-
configuration.setBoolean(LeaderElectionContext.LEADER_IN_TEST, true);
44+
MetastoreConf.setLongVar(configuration, MetastoreConf.ConfVars.HMS_HANDLER_ATTEMPTS, 3);
45+
MetastoreConf.setTimeVar(configuration, MetastoreConf.ConfVars.HMS_HANDLER_INTERVAL, 100, TimeUnit.MILLISECONDS);
4846
configuration.set("hive.txn.manager", "org.apache.hadoop.hive.ql.lockmgr.DbTxnManager");
49-
47+
LeaderElectionFactory.addElectionCreator(LeaderElectionFactory.Method.LOCK, conf -> new ReleaseAndRequireLease(conf, false));
5048
setup(null, configuration);
5149

5250
Configuration conf = new Configuration(configuration);
53-
conf.setBoolean(LeaseLeaderElection.METASTORE_RENEW_LEASE, true);
54-
election = new LeaseLeaderElection();
55-
election.setName("TestMetastoreLeaseLeader");
56-
TableName tableName = (TableName) LeaderElectionContext.getLeaderMutex(conf,
57-
LeaderElectionContext.TTYPE.HOUSEKEEPING, null);
58-
election.tryBeLeader(conf, tableName);
51+
elector = new CombinedLeaderElector(conf);
52+
elector.setName("TestMetastoreLeaseLeader");
53+
elector.tryBeLeader();
5954
}
6055

6156
@Test
6257
public void testHouseKeepingThreads() throws Exception {
63-
CountDownLatch latch = new CountDownLatch(1);
64-
MetastoreHousekeepingLeaderTestBase.TestLeaderNotification.setMonitor(latch);
58+
int size = LeaderElectionContext.TTYPE.values().length;
59+
CountDownLatch latch = new CountDownLatch(size);
60+
MetastoreHousekeepingLeaderTestBase.ReleaseAndRequireLease.setMonitor(latch);
6561
// hms is the leader now
6662
checkHouseKeepingThreadExistence(true);
67-
assertFalse(election.isLeader());
63+
assertFalse(elector.isLeader());
6864
latch.await();
69-
// the lease of hms is timeout, election becomes leader now
70-
assertTrue(election.isLeader());
71-
// hms should shutdown all housekeeping tasks
65+
// the lease of hms is timeout, the elector becomes leader now
66+
assertTrue(elector.isLeader());
67+
// hms should shut down all housekeeping tasks
7268
checkHouseKeepingThreadExistence(false);
7369

74-
latch = new CountDownLatch(1);
75-
MetastoreHousekeepingLeaderTestBase.TestLeaderNotification.setMonitor(latch);
76-
election.close();
70+
latch = new CountDownLatch(size);
71+
MetastoreHousekeepingLeaderTestBase.ReleaseAndRequireLease.setMonitor(latch);
72+
elector.close();
7773
latch.await();
7874
// hms becomes leader again
7975
checkHouseKeepingThreadExistence(true);
8076
}
8177

8278
@After
8379
public void afterTest() {
84-
MetastoreHousekeepingLeaderTestBase.TestLeaderNotification.reset();
80+
MetastoreHousekeepingLeaderTestBase.ReleaseAndRequireLease.reset();
8581
}
8682

8783
}

‎itests/hive-unit/src/test/java/org/apache/hadoop/hive/metastore/TestMetastoreLeaseNonLeader.java‎

Lines changed: 12 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,9 @@
1919
package org.apache.hadoop.hive.metastore;
2020

2121
import org.apache.hadoop.conf.Configuration;
22-
import org.apache.hadoop.hive.common.TableName;
2322
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
24-
import org.apache.hadoop.hive.metastore.leader.LeaderElection;
2523
import org.apache.hadoop.hive.metastore.leader.LeaderElectionContext;
26-
import org.apache.hadoop.hive.metastore.leader.LeaseLeaderElection;
24+
import org.apache.hadoop.hive.metastore.leader.LeaderElectionFactory;
2725
import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
2826
import org.junit.After;
2927
import org.junit.Before;
@@ -36,42 +34,39 @@
3634

3735
public class TestMetastoreLeaseNonLeader extends MetastoreHousekeepingLeaderTestBase {
3836

39-
LeaderElection<TableName> election;
37+
CombinedLeaderElector elector;
4038

4139
@Before
4240
public void setUp() throws Exception {
4341
Configuration conf = MetastoreConf.newMetastoreConf();
4442
TestTxnDbUtil.setConfValues(conf);
4543
TestTxnDbUtil.prepDb(conf);
46-
election = new LeaseLeaderElection();
47-
election.setName("TestMetastoreLeaseNonLeader");
48-
MetastoreConf.setVar(conf, MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_LEADER_ELECTION, "lock");
49-
TableName tableName = (TableName) LeaderElectionContext.getLeaderMutex(conf,
50-
LeaderElectionContext.TTYPE.HOUSEKEEPING, null);
51-
election.tryBeLeader(conf, tableName);
52-
assertTrue("The elector should hold the lease now", election.isLeader());
44+
elector = new CombinedLeaderElector(conf);
45+
elector.setName("TestMetastoreLeaseNonLeader");
46+
elector.tryBeLeader();
47+
assertTrue("The elector should hold the lease now", elector.isLeader());
5348
// start the non-leader hms now
5449
Configuration configuration = new Configuration(conf);
55-
MetastoreConf.setTimeVar(configuration, MetastoreConf.ConfVars.LOCK_SLEEP_BETWEEN_RETRIES, 1, TimeUnit.SECONDS);
56-
configuration.setBoolean(LeaderElectionContext.LEADER_IN_TEST, true);
50+
MetastoreConf.setTimeVar(configuration, MetastoreConf.ConfVars.LOCK_SLEEP_BETWEEN_RETRIES, 100, TimeUnit.MILLISECONDS);
51+
LeaderElectionFactory.addElectionCreator(LeaderElectionFactory.Method.LOCK, c -> new ReleaseAndRequireLease(c, true));
5752
setup(null, configuration);
5853
}
5954

6055
@Test
6156
public void testHouseKeepingThreads() throws Exception {
6257
checkHouseKeepingThreadExistence(false);
6358
// elector releases the lease
64-
CountDownLatch latch = new CountDownLatch(1);
65-
MetastoreHousekeepingLeaderTestBase.TestLeaderNotification.setMonitor(latch);
66-
election.close();
59+
CountDownLatch latch = new CountDownLatch(LeaderElectionContext.TTYPE.values().length);
60+
MetastoreHousekeepingLeaderTestBase.ReleaseAndRequireLease.setMonitor(latch);
61+
elector.close();
6762
latch.await();
6863
// housing threads are here now as the hms wins the election
6964
checkHouseKeepingThreadExistence(true);
7065
}
7166

7267
@After
7368
public void afterTest() {
74-
MetastoreHousekeepingLeaderTestBase.TestLeaderNotification.reset();
69+
MetastoreHousekeepingLeaderTestBase.ReleaseAndRequireLease.reset();
7570
}
7671

7772
}

0 commit comments

Comments
 (0)