Skip to content

Commit 6264c3e

Browse files
KeeProMiseKeeProMise
committed
use RouterAsyncProtocolTestBase
1 parent 915fa05 commit 6264c3e

File tree

4 files changed

+30
-416
lines changed

4 files changed

+30
-416
lines changed

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/TestRouterAsyncCacheAdmin.java

Lines changed: 7 additions & 102 deletions
Original file line numberDiff line numberDiff line change
@@ -18,132 +18,37 @@
1818

1919
package org.apache.hadoop.hdfs.server.federation.router.async;
2020

21-
import org.apache.hadoop.conf.Configuration;
2221
import org.apache.hadoop.fs.CacheFlag;
2322
import org.apache.hadoop.fs.FSDataOutputStream;
24-
import org.apache.hadoop.fs.FileSystem;
2523
import org.apache.hadoop.fs.Path;
2624
import org.apache.hadoop.hdfs.protocol.CacheDirectiveEntry;
2725
import org.apache.hadoop.hdfs.protocol.CacheDirectiveInfo;
2826
import org.apache.hadoop.hdfs.protocol.CachePoolEntry;
2927
import org.apache.hadoop.hdfs.protocol.CachePoolInfo;
30-
import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster;
31-
import org.apache.hadoop.hdfs.server.federation.MockResolver;
32-
import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
3328
import org.apache.hadoop.fs.BatchedRemoteIterator.BatchedEntries;
34-
import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
35-
import org.apache.hadoop.hdfs.server.federation.router.RouterRpcServer;
36-
import org.apache.hadoop.ipc.CallerContext;
37-
import org.junit.After;
38-
import org.junit.AfterClass;
3929
import org.junit.Before;
40-
import org.junit.BeforeClass;
4130
import org.junit.Test;
42-
import org.mockito.Mockito;
4331

4432
import java.io.IOException;
4533
import java.util.EnumSet;
46-
import java.util.concurrent.TimeUnit;
47-
48-
import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
49-
import static org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.DEFAULT_HEARTBEAT_INTERVAL_MS;
50-
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_RPC_ASYNC_HANDLER_COUNT;
51-
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_RPC_ASYNC_RESPONDER_COUNT;
5234
import static org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil.syncReturn;
5335
import static org.junit.Assert.assertEquals;
54-
import static org.junit.Assert.assertTrue;
55-
56-
public class TestRouterAsyncCacheAdmin {
57-
private static Configuration routerConf;
58-
/** Federated HDFS cluster. */
59-
private static MiniRouterDFSCluster cluster;
60-
private static String ns0;
6136

62-
/** Random Router for this federated cluster. */
63-
private MiniRouterDFSCluster.RouterContext router;
64-
private FileSystem routerFs;
65-
private RouterRpcServer routerRpcServer;
37+
/**
38+
* Used to test the functionality of {@link RouterAsyncCacheAdmin}.
39+
*/
40+
public class TestRouterAsyncCacheAdmin extends RouterAsyncProtocolTestBase {
6641
private RouterAsyncCacheAdmin asyncCacheAdmin;
6742

68-
@BeforeClass
69-
public static void setUpCluster() throws Exception {
70-
cluster = new MiniRouterDFSCluster(true, 1, 2,
71-
DEFAULT_HEARTBEAT_INTERVAL_MS, 1000);
72-
cluster.setNumDatanodesPerNameservice(3);
73-
74-
cluster.startCluster();
75-
76-
// Making one Namenode active per nameservice
77-
if (cluster.isHighAvailability()) {
78-
for (String ns : cluster.getNameservices()) {
79-
cluster.switchToActive(ns, NAMENODES[0]);
80-
cluster.switchToStandby(ns, NAMENODES[1]);
81-
}
82-
}
83-
// Start routers with only an RPC service
84-
routerConf = new RouterConfigBuilder()
85-
.rpc()
86-
.build();
87-
88-
// Reduce the number of RPC clients threads to overload the Router easy
89-
routerConf.setInt(RBFConfigKeys.DFS_ROUTER_CLIENT_THREADS_SIZE, 1);
90-
routerConf.setInt(DFS_ROUTER_RPC_ASYNC_HANDLER_COUNT, 1);
91-
routerConf.setInt(DFS_ROUTER_RPC_ASYNC_RESPONDER_COUNT, 1);
92-
// We decrease the DN cache times to make the test faster
93-
routerConf.setTimeDuration(
94-
RBFConfigKeys.DN_REPORT_CACHE_EXPIRE, 1, TimeUnit.SECONDS);
95-
cluster.addRouterOverrides(routerConf);
96-
// Start routers with only an RPC service
97-
cluster.startRouters();
98-
99-
// Register and verify all NNs with all routers
100-
cluster.registerNamenodes();
101-
cluster.waitNamenodeRegistration();
102-
cluster.waitActiveNamespaces();
103-
ns0 = cluster.getNameservices().get(0);
104-
}
105-
106-
@AfterClass
107-
public static void shutdownCluster() throws Exception {
108-
if (cluster != null) {
109-
cluster.shutdown();
110-
}
111-
}
112-
11343
@Before
114-
public void setUp() throws IOException {
115-
router = cluster.getRandomRouter();
116-
routerFs = router.getFileSystem();
117-
routerRpcServer = router.getRouterRpcServer();
118-
routerRpcServer.initAsyncThreadPool();
119-
RouterAsyncRpcClient asyncRpcClient = new RouterAsyncRpcClient(
120-
routerConf, router.getRouter(), routerRpcServer.getNamenodeResolver(),
121-
routerRpcServer.getRPCMonitor(),
122-
routerRpcServer.getRouterStateIdContext());
123-
RouterRpcServer spy = Mockito.spy(routerRpcServer);
124-
Mockito.when(spy.getRPCClient()).thenReturn(asyncRpcClient);
125-
asyncCacheAdmin = new RouterAsyncCacheAdmin(spy);
126-
127-
// Create mock locations
128-
MockResolver resolver = (MockResolver) router.getRouter().getSubclusterResolver();
129-
resolver.addLocation("/", ns0, "/");
130-
FSDataOutputStream fsDataOutputStream = routerFs.create(
44+
public void setup() throws IOException {
45+
asyncCacheAdmin = new RouterAsyncCacheAdmin(getRouterAsyncRpcServer());
46+
FSDataOutputStream fsDataOutputStream = getRouterFs().create(
13147
new Path("/testCache.file"), true);
13248
fsDataOutputStream.write(new byte[1024]);
13349
fsDataOutputStream.close();
13450
}
13551

136-
@After
137-
public void tearDown() throws IOException {
138-
// clear client context
139-
CallerContext.setCurrent(null);
140-
boolean delete = routerFs.delete(new Path("/testCache.file"));
141-
assertTrue(delete);
142-
if (routerFs != null) {
143-
routerFs.close();
144-
}
145-
}
146-
14752
@Test
14853
public void testRouterAsyncCacheAdmin() throws Exception {
14954
asyncCacheAdmin.addCachePool(new CachePoolInfo("pool"));

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/async/TestRouterAsyncRpcServer.java

Lines changed: 5 additions & 103 deletions
Original file line numberDiff line numberDiff line change
@@ -17,131 +17,33 @@
1717
*/
1818
package org.apache.hadoop.hdfs.server.federation.router.async;
1919

20-
import org.apache.hadoop.conf.Configuration;
21-
import org.apache.hadoop.fs.FileSystem;
22-
import org.apache.hadoop.fs.Path;
23-
import org.apache.hadoop.fs.permission.FsPermission;
2420
import org.apache.hadoop.hdfs.protocol.BlockStoragePolicy;
2521
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
2622
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
27-
import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster;
28-
import org.apache.hadoop.hdfs.server.federation.MockResolver;
29-
import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
3023
import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
31-
import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
3224
import org.apache.hadoop.hdfs.server.federation.router.RemoteMethod;
3325
import org.apache.hadoop.hdfs.server.federation.router.RouterRpcServer;
3426
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorageReport;
35-
import org.apache.hadoop.ipc.CallerContext;
36-
import org.junit.After;
37-
import org.junit.AfterClass;
3827
import org.junit.Before;
39-
import org.junit.BeforeClass;
4028
import org.junit.Test;
41-
import org.mockito.Mockito;
4229

4330
import java.io.IOException;
4431
import java.util.List;
4532
import java.util.Map;
46-
import java.util.concurrent.TimeUnit;
4733

48-
import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
49-
import static org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.DEFAULT_HEARTBEAT_INTERVAL_MS;
50-
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_RPC_ASYNC_HANDLER_COUNT;
51-
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_RPC_ASYNC_RESPONDER_COUNT;
5234
import static org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil.syncReturn;
5335
import static org.junit.Assert.assertEquals;
5436
import static org.junit.Assert.assertNotNull;
55-
import static org.junit.Assert.assertTrue;
5637

5738
/**
5839
* Used to test the async functionality of {@link RouterRpcServer}.
5940
*/
60-
public class TestRouterAsyncRpcServer {
61-
private static Configuration routerConf;
62-
/** Federated HDFS cluster. */
63-
private static MiniRouterDFSCluster cluster;
64-
private static String ns0;
65-
66-
/** Random Router for this federated cluster. */
67-
private MiniRouterDFSCluster.RouterContext router;
68-
private FileSystem routerFs;
41+
public class TestRouterAsyncRpcServer extends RouterAsyncProtocolTestBase {
6942
private RouterRpcServer asyncRouterRpcServer;
7043

71-
@BeforeClass
72-
public static void setUpCluster() throws Exception {
73-
cluster = new MiniRouterDFSCluster(true, 1, 2,
74-
DEFAULT_HEARTBEAT_INTERVAL_MS, 1000);
75-
cluster.setNumDatanodesPerNameservice(3);
76-
77-
cluster.startCluster();
78-
79-
// Making one Namenode active per nameservice
80-
if (cluster.isHighAvailability()) {
81-
for (String ns : cluster.getNameservices()) {
82-
cluster.switchToActive(ns, NAMENODES[0]);
83-
cluster.switchToStandby(ns, NAMENODES[1]);
84-
}
85-
}
86-
// Start routers with only an RPC service
87-
routerConf = new RouterConfigBuilder()
88-
.rpc()
89-
.build();
90-
91-
// Reduce the number of RPC clients threads to overload the Router easy
92-
routerConf.setInt(RBFConfigKeys.DFS_ROUTER_CLIENT_THREADS_SIZE, 1);
93-
routerConf.setInt(DFS_ROUTER_RPC_ASYNC_HANDLER_COUNT, 1);
94-
routerConf.setInt(DFS_ROUTER_RPC_ASYNC_RESPONDER_COUNT, 1);
95-
// We decrease the DN cache times to make the test faster
96-
routerConf.setTimeDuration(
97-
RBFConfigKeys.DN_REPORT_CACHE_EXPIRE, 1, TimeUnit.SECONDS);
98-
cluster.addRouterOverrides(routerConf);
99-
// Start routers with only an RPC service
100-
cluster.startRouters();
101-
102-
// Register and verify all NNs with all routers
103-
cluster.registerNamenodes();
104-
cluster.waitNamenodeRegistration();
105-
cluster.waitActiveNamespaces();
106-
ns0 = cluster.getNameservices().get(0);
107-
}
108-
109-
@AfterClass
110-
public static void shutdownCluster() throws Exception {
111-
if (cluster != null) {
112-
cluster.shutdown();
113-
}
114-
}
115-
11644
@Before
117-
public void setUp() throws IOException {
118-
router = cluster.getRandomRouter();
119-
routerFs = router.getFileSystem();
120-
RouterRpcServer routerRpcServer = router.getRouterRpcServer();
121-
routerRpcServer.initAsyncThreadPool();
122-
RouterAsyncRpcClient asyncRpcClient = new RouterAsyncRpcClient(
123-
routerConf, router.getRouter(), routerRpcServer.getNamenodeResolver(),
124-
routerRpcServer.getRPCMonitor(),
125-
routerRpcServer.getRouterStateIdContext());
126-
asyncRouterRpcServer = Mockito.spy(routerRpcServer);
127-
Mockito.when(asyncRouterRpcServer.getRPCClient()).thenReturn(asyncRpcClient);
128-
129-
// Create mock locations
130-
MockResolver resolver = (MockResolver) router.getRouter().getSubclusterResolver();
131-
resolver.addLocation("/", ns0, "/");
132-
FsPermission permission = new FsPermission("705");
133-
routerFs.mkdirs(new Path("/testdir"), permission);
134-
}
135-
136-
@After
137-
public void tearDown() throws IOException {
138-
// clear client context
139-
CallerContext.setCurrent(null);
140-
boolean delete = routerFs.delete(new Path("/testdir"));
141-
assertTrue(delete);
142-
if (routerFs != null) {
143-
routerFs.close();
144-
}
45+
public void setup() throws IOException {
46+
asyncRouterRpcServer = getRouterAsyncRpcServer();
14547
}
14648

14749
/**
@@ -165,7 +67,7 @@ public void testGetCreateLocationAsync() throws Exception {
16567
asyncRouterRpcServer.getCreateLocationAsync("/testdir", locations);
16668
RemoteLocation remoteLocation = syncReturn(RemoteLocation.class);
16769
assertNotNull(remoteLocation);
168-
assertEquals(ns0, remoteLocation.getNameserviceId());
70+
assertEquals(getNs0(), remoteLocation.getNameserviceId());
16971
}
17072

17173
/**
@@ -182,7 +84,7 @@ public void testGetDatanodeReportAsync() throws Exception {
18284
asyncRouterRpcServer.getDatanodeStorageReportMapAsync(HdfsConstants.DatanodeReportType.ALL);
18385
Map<String, DatanodeStorageReport[]> map = syncReturn(Map.class);
18486
assertEquals(1, map.size());
185-
assertEquals(3, map.get(ns0).length);
87+
assertEquals(3, map.get(getNs0()).length);
18688

18789
DatanodeInfo[] slowDatanodeReport1 =
18890
asyncRouterRpcServer.getSlowDatanodeReport(true, 0);

0 commit comments

Comments
 (0)