Apache/hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-12919. RBF: Support erasure coding methods in RouterRpcServer. Contributed by Inigo Goiri.

Browse Source
This commit is contained in:
Inigo Goiri 2018-03-15 11:30:52 -07:00
parent ccf07fdafd
commit b1b10007a4
5 changed files with 136 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Quota.java
View File

@ -94,8 +94,8 @@ public QuotaUsage getQuotaUsage(String path) throws IOException {
final List<RemoteLocation> quotaLocs = getValidQuotaLocations(path); final List<RemoteLocation> quotaLocs = getValidQuotaLocations(path);
RemoteMethod method = new RemoteMethod("getQuotaUsage", RemoteMethod method = new RemoteMethod("getQuotaUsage",
new Class<?>[] {String.class}, new RemoteParam()); new Class<?>[] {String.class}, new RemoteParam());
Map<RemoteLocation, Object> results = rpcClient.invokeConcurrent(quotaLocs, Map<RemoteLocation, QuotaUsage> results = rpcClient.invokeConcurrent(
method, true, false); quotaLocs, method, true, false, QuotaUsage.class);
return aggregateQuota(results); return aggregateQuota(results);
} }
@ -151,14 +151,14 @@ private List<RemoteLocation> getValidQuotaLocations(String path)
* @param results Quota query result. * @param results Quota query result.
* @return Aggregated Quota. * @return Aggregated Quota.
*/ */
private QuotaUsage aggregateQuota(Map<RemoteLocation, Object> results) { private QuotaUsage aggregateQuota(Map<RemoteLocation, QuotaUsage> results) {
long nsCount = 0; long nsCount = 0;
long ssCount = 0; long ssCount = 0;
boolean hasQuotaUnSet = false; boolean hasQuotaUnSet = false;
for (Map.Entry<RemoteLocation, Object> entry : results.entrySet()) { for (Map.Entry<RemoteLocation, QuotaUsage> entry : results.entrySet()) {
RemoteLocation loc = entry.getKey(); RemoteLocation loc = entry.getKey();
QuotaUsage usage = (QuotaUsage) entry.getValue(); QuotaUsage usage = entry.getValue();
if (usage != null) { if (usage != null) {
// If quota is not set in real FileSystem, the usage // If quota is not set in real FileSystem, the usage
// value will return -1. // value will return -1.

63
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java
View File

@ -666,9 +666,9 @@ public Object invokeSequential(
* @throws IOException if the success condition is not met, return the first * @throws IOException if the success condition is not met, return the first
* remote exception generated. * remote exception generated.
*/ */
public Object invokeSequential( public <T> T invokeSequential(
final List<? extends RemoteLocationContext> locations, final List<? extends RemoteLocationContext> locations,
final RemoteMethod remoteMethod, Class<?> expectedResultClass, final RemoteMethod remoteMethod, Class<T> expectedResultClass,
Object expectedResultValue) throws IOException { Object expectedResultValue) throws IOException {
final UserGroupInformation ugi = RouterRpcServer.getRemoteUser(); final UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
@ -688,7 +688,9 @@ public Object invokeSequential(
if (isExpectedClass(expectedResultClass, result) && if (isExpectedClass(expectedResultClass, result) &&
isExpectedValue(expectedResultValue, result)) { isExpectedValue(expectedResultValue, result)) {
// Valid result, stop here // Valid result, stop here
return result; @SuppressWarnings("unchecked")
T ret = (T)result;
return ret;
} }
if (firstResult == null) { if (firstResult == null) {
firstResult = result; firstResult = result;
@ -718,7 +720,9 @@ public Object invokeSequential(
throw firstThrownException; throw firstThrownException;
} }
// Return the last result, whether it is the value we are looking for or a // Return the last result, whether it is the value we are looking for or a
return firstResult; @SuppressWarnings("unchecked")
T ret = (T)firstResult;
return ret;
} }
/** /**
@ -758,7 +762,7 @@ private static boolean isExpectedValue(Object expectedValue, Object value) {
} }
/** /**
* Invokes multiple concurrent proxy calls to different clients. Returns an * Invoke multiple concurrent proxy calls to different clients. Returns an
* array of results. * array of results.
* *
* Re-throws exceptions generated by the remote RPC call as either * Re-throws exceptions generated by the remote RPC call as either
@ -771,14 +775,12 @@ private static boolean isExpectedValue(Object expectedValue, Object value) {
* not complete. If false exceptions are ignored and all data results * not complete. If false exceptions are ignored and all data results
* successfully received are returned. * successfully received are returned.
* @param standby If the requests should go to the standby namenodes too. * @param standby If the requests should go to the standby namenodes too.
* @return Result of invoking the method per subcluster: nsId -> result. * @throws IOException If all the calls throw an exception.
* @throws IOException If requiredResponse=true and any of the calls throw an
* exception.
*/ */
public <T extends RemoteLocationContext> Map<T, Object> invokeConcurrent( public <T extends RemoteLocationContext, R> void invokeConcurrent(
final Collection<T> locations, final RemoteMethod method, final Collection<T> locations, final RemoteMethod method,
boolean requireResponse, boolean standby) throws IOException { boolean requireResponse, boolean standby) throws IOException {
return invokeConcurrent(locations, method, requireResponse, standby, -1); invokeConcurrent(locations, method, requireResponse, standby, void.class);
} }
/** /**
@ -788,6 +790,36 @@ public <T extends RemoteLocationContext> Map<T, Object> invokeConcurrent(
* Re-throws exceptions generated by the remote RPC call as either * Re-throws exceptions generated by the remote RPC call as either
* RemoteException or IOException. * RemoteException or IOException.
* *
* @param <T> The type of the remote location.
* @param <R> The type of the remote method return.
* @param locations List of remote locations to call concurrently.
* @param method The remote method and parameters to invoke.
* @param requireResponse If true an exception will be thrown if all calls do
* not complete. If false exceptions are ignored and all data results
* successfully received are returned.
* @param standby If the requests should go to the standby namenodes too.
* @param clazz Type of the remote return type.
* @return Result of invoking the method per subcluster: nsId -> result.
* @throws IOException If requiredResponse=true and any of the calls throw an
* exception.
*/
public <T extends RemoteLocationContext, R> Map<T, R> invokeConcurrent(
final Collection<T> locations, final RemoteMethod method,
boolean requireResponse, boolean standby, Class<R> clazz)
throws IOException {
return invokeConcurrent(
locations, method, requireResponse, standby, -1, clazz);
}
/**
* Invokes multiple concurrent proxy calls to different clients. Returns an
* array of results.
*
* Re-throws exceptions generated by the remote RPC call as either
* RemoteException or IOException.
*
* @param <T> The type of the remote location.
* @param <R> The type of the remote method return.
* @param locations List of remote locations to call concurrently. * @param locations List of remote locations to call concurrently.
* @param method The remote method and parameters to invoke. * @param method The remote method and parameters to invoke.
* @param requireResponse If true an exception will be thrown if all calls do * @param requireResponse If true an exception will be thrown if all calls do
@ -795,14 +827,15 @@ public <T extends RemoteLocationContext> Map<T, Object> invokeConcurrent(
* successfully received are returned. * successfully received are returned.
* @param standby If the requests should go to the standby namenodes too. * @param standby If the requests should go to the standby namenodes too.
* @param timeOutMs Timeout for each individual call. * @param timeOutMs Timeout for each individual call.
* @param clazz Type of the remote return type.
* @return Result of invoking the method per subcluster: nsId -> result. * @return Result of invoking the method per subcluster: nsId -> result.
* @throws IOException If requiredResponse=true and any of the calls throw an * @throws IOException If requiredResponse=true and any of the calls throw an
* exception. * exception.
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public <T extends RemoteLocationContext> Map<T, Object> invokeConcurrent( public <T extends RemoteLocationContext, R> Map<T, R> invokeConcurrent(
final Collection<T> locations, final RemoteMethod method, final Collection<T> locations, final RemoteMethod method,
boolean requireResponse, boolean standby, long timeOutMs) boolean requireResponse, boolean standby, long timeOutMs, Class<R> clazz)
throws IOException { throws IOException {
final UserGroupInformation ugi = RouterRpcServer.getRemoteUser(); final UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
@ -816,7 +849,7 @@ public <T extends RemoteLocationContext> Map<T, Object> invokeConcurrent(
getNamenodesForNameservice(ns); getNamenodesForNameservice(ns);
Object[] paramList = method.getParams(location); Object[] paramList = method.getParams(location);
Object result = invokeMethod(ugi, namenodes, m, paramList); Object result = invokeMethod(ugi, namenodes, m, paramList);
return Collections.singletonMap(location, result); return Collections.singletonMap(location, clazz.cast(result));
} }
List<T> orderedLocations = new LinkedList<>(); List<T> orderedLocations = new LinkedList<>();
@ -866,14 +899,14 @@ public Object call() throws Exception {
} else { } else {
futures = executorService.invokeAll(callables); futures = executorService.invokeAll(callables);
} }
Map<T, Object> results = new TreeMap<>(); Map<T, R> results = new TreeMap<>();
Map<T, IOException> exceptions = new TreeMap<>(); Map<T, IOException> exceptions = new TreeMap<>();
for (int i=0; i<futures.size(); i++) { for (int i=0; i<futures.size(); i++) {
T location = orderedLocations.get(i); T location = orderedLocations.get(i);
try { try {
Future<Object> future = futures.get(i); Future<Object> future = futures.get(i);
Object result = future.get(); Object result = future.get();
results.put(location, result); results.put(location, clazz.cast(result));
} catch (CancellationException ce) { } catch (CancellationException ce) {
T loc = orderedLocations.get(i); T loc = orderedLocations.get(i);
String msg = String msg =

127
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
View File

@ -28,12 +28,14 @@
import java.io.FileNotFoundException; import java.io.FileNotFoundException;
import java.io.IOException; import java.io.IOException;
import java.lang.reflect.Array;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.util.Collection; import java.util.Collection;
import java.util.EnumSet; import java.util.EnumSet;
import java.util.HashMap; import java.util.HashMap;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -966,8 +968,9 @@ public DirectoryListing getListing(String src, byte[] startAfter,
RemoteMethod method = new RemoteMethod("getListing", RemoteMethod method = new RemoteMethod("getListing",
new Class<?>[] {String.class, startAfter.getClass(), boolean.class}, new Class<?>[] {String.class, startAfter.getClass(), boolean.class},
new RemoteParam(), startAfter, needLocation); new RemoteParam(), startAfter, needLocation);
Map<RemoteLocation, Object> listings = Map<RemoteLocation, DirectoryListing> listings =
rpcClient.invokeConcurrent(locations, method, false, false); rpcClient.invokeConcurrent(
locations, method, false, false, DirectoryListing.class);
Map<String, HdfsFileStatus> nnListing = new TreeMap<>(); Map<String, HdfsFileStatus> nnListing = new TreeMap<>();
int totalRemainingEntries = 0; int totalRemainingEntries = 0;
@ -976,9 +979,10 @@ public DirectoryListing getListing(String src, byte[] startAfter,
if (listings != null) { if (listings != null) {
// Check the subcluster listing with the smallest name // Check the subcluster listing with the smallest name
String lastName = null; String lastName = null;
for (Entry<RemoteLocation, Object> entry : listings.entrySet()) { for (Entry<RemoteLocation, DirectoryListing> entry :
listings.entrySet()) {
RemoteLocation location = entry.getKey(); RemoteLocation location = entry.getKey();
DirectoryListing listing = (DirectoryListing) entry.getValue(); DirectoryListing listing = entry.getValue();
if (listing == null) { if (listing == null) {
LOG.debug("Cannot get listing from {}", location); LOG.debug("Cannot get listing from {}", location);
} else { } else {
@ -1102,11 +1106,10 @@ public long[] getStats() throws IOException {
RemoteMethod method = new RemoteMethod("getStats"); RemoteMethod method = new RemoteMethod("getStats");
Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> results = Map<FederationNamespaceInfo, long[]> results =
rpcClient.invokeConcurrent(nss, method, true, false); rpcClient.invokeConcurrent(nss, method, true, false, long[].class);
long[] combinedData = new long[STATS_ARRAY_LENGTH]; long[] combinedData = new long[STATS_ARRAY_LENGTH];
for (Object o : results.values()) { for (long[] data : results.values()) {
long[] data = (long[]) o;
for (int i = 0; i < combinedData.length && i < data.length; i++) { for (int i = 0; i < combinedData.length && i < data.length; i++) {
if (data[i] >= 0) { if (data[i] >= 0) {
combinedData[i] += data[i]; combinedData[i] += data[i];
@ -1139,11 +1142,13 @@ public DatanodeInfo[] getDatanodeReport(
new Class<?>[] {DatanodeReportType.class}, type); new Class<?>[] {DatanodeReportType.class}, type);
Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> results = Map<FederationNamespaceInfo, DatanodeInfo[]> results =
rpcClient.invokeConcurrent(nss, method, true, false, timeOutMs); rpcClient.invokeConcurrent(
for (Entry<FederationNamespaceInfo, Object> entry : results.entrySet()) { nss, method, true, false, timeOutMs, DatanodeInfo[].class);
for (Entry<FederationNamespaceInfo, DatanodeInfo[]> entry :
results.entrySet()) {
FederationNamespaceInfo ns = entry.getKey(); FederationNamespaceInfo ns = entry.getKey();
DatanodeInfo[] result = (DatanodeInfo[]) entry.getValue(); DatanodeInfo[] result = entry.getValue();
for (DatanodeInfo node : result) { for (DatanodeInfo node : result) {
String nodeId = node.getXferAddr(); String nodeId = node.getXferAddr();
if (!datanodesMap.containsKey(nodeId)) { if (!datanodesMap.containsKey(nodeId)) {
@ -1229,17 +1234,14 @@ public boolean setSafeMode(SafeModeAction action, boolean isChecked)
new Class<?>[] {SafeModeAction.class, boolean.class}, new Class<?>[] {SafeModeAction.class, boolean.class},
action, isChecked); action, isChecked);
Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> results = Map<FederationNamespaceInfo, Boolean> results =
rpcClient.invokeConcurrent(nss, method, true, true); rpcClient.invokeConcurrent(nss, method, true, true, boolean.class);
// We only report true if all the name space are in safe mode // We only report true if all the name space are in safe mode
int numSafemode = 0; int numSafemode = 0;
for (Object result : results.values()) { for (boolean safemode : results.values()) {
if (result instanceof Boolean) { if (safemode) {
boolean safemode = (boolean) result; numSafemode++;
if (safemode) {
numSafemode++;
}
} }
} }
return numSafemode == results.size(); return numSafemode == results.size();
@ -1252,18 +1254,14 @@ public boolean restoreFailedStorage(String arg) throws IOException {
RemoteMethod method = new RemoteMethod("restoreFailedStorage", RemoteMethod method = new RemoteMethod("restoreFailedStorage",
new Class<?>[] {String.class}, arg); new Class<?>[] {String.class}, arg);
final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> ret = Map<FederationNamespaceInfo, Boolean> ret =
rpcClient.invokeConcurrent(nss, method, true, false); rpcClient.invokeConcurrent(nss, method, true, false, boolean.class);
boolean success = true; boolean success = true;
Object obj = ret; for (boolean s : ret.values()) {
@SuppressWarnings("unchecked")
Map<FederationNamespaceInfo, Boolean> results =
(Map<FederationNamespaceInfo, Boolean>)obj;
Collection<Boolean> sucesses = results.values();
for (boolean s : sucesses) {
if (!s) { if (!s) {
success = false; success = false;
break;
} }
} }
return success; return success;
@ -1284,17 +1282,12 @@ public long rollEdits() throws IOException {
RemoteMethod method = new RemoteMethod("rollEdits", new Class<?>[] {}); RemoteMethod method = new RemoteMethod("rollEdits", new Class<?>[] {});
final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> ret = Map<FederationNamespaceInfo, Long> ret =
rpcClient.invokeConcurrent(nss, method, true, false); rpcClient.invokeConcurrent(nss, method, true, false, long.class);
// Return the maximum txid // Return the maximum txid
long txid = 0; long txid = 0;
Object obj = ret; for (long t : ret.values()) {
@SuppressWarnings("unchecked")
Map<FederationNamespaceInfo, Long> results =
(Map<FederationNamespaceInfo, Long>)obj;
Collection<Long> txids = results.values();
for (long t : txids) {
if (t > txid) { if (t > txid) {
txid = t; txid = t;
} }
@ -1329,17 +1322,13 @@ public RollingUpgradeInfo rollingUpgrade(RollingUpgradeAction action)
RemoteMethod method = new RemoteMethod("rollingUpgrade", RemoteMethod method = new RemoteMethod("rollingUpgrade",
new Class<?>[] {RollingUpgradeAction.class}, action); new Class<?>[] {RollingUpgradeAction.class}, action);
final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> ret = Map<FederationNamespaceInfo, RollingUpgradeInfo> ret =
rpcClient.invokeConcurrent(nss, method, true, false); rpcClient.invokeConcurrent(
nss, method, true, false, RollingUpgradeInfo.class);
// Return the first rolling upgrade info // Return the first rolling upgrade info
RollingUpgradeInfo info = null; RollingUpgradeInfo info = null;
Object obj = ret; for (RollingUpgradeInfo infoNs : ret.values()) {
@SuppressWarnings("unchecked")
Map<FederationNamespaceInfo, RollingUpgradeInfo> results =
(Map<FederationNamespaceInfo, RollingUpgradeInfo>)obj;
Collection<RollingUpgradeInfo> infos = results.values();
for (RollingUpgradeInfo infoNs : infos) {
if (info == null && infoNs != null) { if (info == null && infoNs != null) {
info = infoNs; info = infoNs;
} }
@ -1391,10 +1380,9 @@ public ContentSummary getContentSummary(String path) throws IOException {
final List<RemoteLocation> locations = getLocationsForPath(path, false); final List<RemoteLocation> locations = getLocationsForPath(path, false);
RemoteMethod method = new RemoteMethod("getContentSummary", RemoteMethod method = new RemoteMethod("getContentSummary",
new Class<?>[] {String.class}, new RemoteParam()); new Class<?>[] {String.class}, new RemoteParam());
@SuppressWarnings("unchecked") Map<RemoteLocation, ContentSummary> results =
Map<String, ContentSummary> results = rpcClient.invokeConcurrent(
(Map<String, ContentSummary>) ((Object)rpcClient.invokeConcurrent( locations, method, false, false, ContentSummary.class);
locations, method, false, false));
summaries.addAll(results.values()); summaries.addAll(results.values());
} catch (FileNotFoundException e) { } catch (FileNotFoundException e) {
notFoundException = e; notFoundException = e;
@ -1767,16 +1755,12 @@ public long getCurrentEditLogTxid() throws IOException {
RemoteMethod method = new RemoteMethod( RemoteMethod method = new RemoteMethod(
"getCurrentEditLogTxid", new Class<?>[] {}); "getCurrentEditLogTxid", new Class<?>[] {});
final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces(); final Set<FederationNamespaceInfo> nss = namenodeResolver.getNamespaces();
Map<FederationNamespaceInfo, Object> ret = Map<FederationNamespaceInfo, Long> ret =
rpcClient.invokeConcurrent(nss, method, true, false); rpcClient.invokeConcurrent(nss, method, true, false, long.class);
// Return the maximum txid // Return the maximum txid
long txid = 0; long txid = 0;
Object obj = ret; Collection<Long> txids = ret.values();
@SuppressWarnings("unchecked")
Map<FederationNamespaceInfo, Long> results =
(Map<FederationNamespaceInfo, Long>)obj;
Collection<Long> txids = results.values();
for (long t : txids) { for (long t : txids) {
if (t > txid) { if (t > txid) {
txid = t; txid = t;
@ -2039,6 +2023,39 @@ static UserGroupInformation getRemoteUser() throws IOException {
return (ugi != null) ? ugi : UserGroupInformation.getCurrentUser(); return (ugi != null) ? ugi : UserGroupInformation.getCurrentUser();
} }
/**
* Merge the outputs from multiple namespaces.
* @param map Namespace -> Output array.
* @param clazz Class of the values.
* @return Array with the outputs.
*/
protected static <T> T[] merge(
Map<FederationNamespaceInfo, T[]> map, Class<T> clazz) {
// Put all results into a set to avoid repeats
Set<T> ret = new LinkedHashSet<>();
for (T[] values : map.values()) {
for (T val : values) {
ret.add(val);
}
}
return toArray(ret, clazz);
}
/**
* Convert a set of values into an array.
* @param set Input set.
* @param clazz Class of the values.
* @return Array with the values in set.
*/
private static <T> T[] toArray(Set<T> set, Class<T> clazz) {
@SuppressWarnings("unchecked")
T[] combinedData = (T[]) Array.newInstance(clazz, set.size());
combinedData = set.toArray(combinedData);
return combinedData;
}
/** /**
* Get quota module implement. * Get quota module implement.
*/ */

12
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/RouterDFSCluster.java
View File

@ -109,6 +109,8 @@ public class RouterDFSCluster {
private List<RouterContext> routers; private List<RouterContext> routers;
/** If the Namenodes are in high availability.*/ /** If the Namenodes are in high availability.*/
private boolean highAvailability; private boolean highAvailability;
/** Number of datanodes per nameservice. */
private int numDatanodesPerNameservice = 2;
/** Mini cluster. */ /** Mini cluster. */
private MiniDFSCluster cluster; private MiniDFSCluster cluster;
@ -356,8 +358,8 @@ public RouterDFSCluster(boolean ha, int numNameservices) {
DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS); DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS);
} }
public RouterDFSCluster(boolean ha, int numNameservices, int numNamnodes) { public RouterDFSCluster(boolean ha, int numNameservices, int numNamenodes) {
this(ha, numNameservices, numNamnodes, this(ha, numNameservices, numNamenodes,
DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS); DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_CACHE_INTERVAL_MS);
} }
@ -531,6 +533,10 @@ public void configureNameservices(int numNameservices, int numNamenodes) {
} }
} }
public void setNumDatanodesPerNameservice(int num) {
this.numDatanodesPerNameservice = num;
}
public String getNameservicesKey() { public String getNameservicesKey() {
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
for (String nsId : this.nameservices) { for (String nsId : this.nameservices) {
@ -658,7 +664,7 @@ public void startCluster(Configuration overrideConf) {
nnConf.addResource(overrideConf); nnConf.addResource(overrideConf);
} }
cluster = new MiniDFSCluster.Builder(nnConf) cluster = new MiniDFSCluster.Builder(nnConf)
.numDataNodes(nameservices.size()*2) .numDataNodes(nameservices.size() * numDatanodesPerNameservice)
.nnTopology(topology) .nnTopology(topology)
.build(); .build();
cluster.waitActive(); cluster.waitActive();

4
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRpc.java
View File

@ -153,9 +153,9 @@ public void testSetup() throws Exception {
// Wait to ensure NN has fully created its test directories // Wait to ensure NN has fully created its test directories
Thread.sleep(100); Thread.sleep(100);
// Pick a NS, namenode and router for this test // Default namenode and random router for this test
this.router = cluster.getRandomRouter(); this.router = cluster.getRandomRouter();
this.ns = cluster.getRandomNameservice(); this.ns = cluster.getNameservices().get(0);
this.namenode = cluster.getNamenode(ns, null); this.namenode = cluster.getNamenode(ns, null);
// Handles to the ClientProtocol interface // Handles to the ClientProtocol interface