HdfsClient extends BaseClient { public static void validateConnectionConfigs(Map<String, String> configs) throws IllegalArgumentException { String lookupPrincipal=null; try{ lookupPrincipal = SecureClientLogin.getPrincipal(configs.get("lookupprincipal"), java.net.InetAddress.getLocalHost().getCanonicalHostName()); }catch(Exception e){ } String lookupKeytab = configs.get("lookupkeytab"); if(StringUtils.isEmpty(lookupPrincipal) || StringUtils.isEmpty(lookupKeytab)){ String username = configs.get("username"); if ((username == null || username.isEmpty())) { throw new IllegalArgumentException("Value for username not specified"); } String password = configs.get("password"); if ((password == null || password.isEmpty())) { throw new IllegalArgumentException("Value for password not specified"); } } String authentication = configs.get("hadoop.security.authentication"); if ((authentication == null || authentication.isEmpty())) { throw new IllegalArgumentException("Value for hadoop.security.authentication not specified"); } String fsDefaultName = configs.get("fs.default.name"); fsDefaultName = (fsDefaultName == null) ? "" : fsDefaultName.trim(); if (fsDefaultName.isEmpty()) { throw new IllegalArgumentException("Value for fs.default.name not specified"); } else { String[] fsDefaultNameElements = fsDefaultName.split(","); for (String fsDefaultNameElement : fsDefaultNameElements) { if (fsDefaultNameElement.isEmpty()) { throw new IllegalArgumentException( "Value for " + "fs.default.name element" + fsDefaultNameElement + " not specified"); } } if (fsDefaultNameElements != null && fsDefaultNameElements.length >= 2) { String cluster = ""; StringBuilder clusters = new StringBuilder(); configs.put("dfs.nameservices", "hdfscluster"); configs.put("fs.default.name", "hdfs: configs.put("dfs.client.failover.proxy.provider." + configs.get("dfs.nameservices"), "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); for (int i = 0; i < fsDefaultNameElements.length; i++) { cluster = "namenode" + (i + 1); configs.put("dfs.namenode.rpc-address." + configs.get("dfs.nameservices") + "." + cluster, fsDefaultNameElements[i]); if (i == (fsDefaultNameElements.length - 1)) { clusters.append(cluster); } else { clusters.append(cluster).append(","); } } configs.put("dfs.ha.namenodes." + configs.get("dfs.nameservices"), clusters.toString()); } } String dfsNameservices = configs.get("dfs.nameservices"); dfsNameservices = (dfsNameservices == null) ? "" : dfsNameservices.trim(); if (!dfsNameservices.isEmpty()) { String[] dfsNameserviceElements = dfsNameservices.split(","); for (String dfsNameserviceElement : dfsNameserviceElements) { String proxyProvider = configs.get("dfs.client.failover.proxy.provider." + dfsNameserviceElement); proxyProvider = (proxyProvider == null) ? "" : proxyProvider.trim(); if (proxyProvider.isEmpty()) { throw new IllegalArgumentException("Value for " + "dfs.client.failover.proxy.provider." + dfsNameserviceElement + " not specified"); } String dfsNameNodes = configs.get("dfs.ha.namenodes." + dfsNameserviceElement); dfsNameNodes = (dfsNameNodes == null) ? "" : dfsNameNodes.trim(); if (dfsNameNodes.isEmpty()) { throw new IllegalArgumentException("Value for " + "dfs.ha.namenodes." + dfsNameserviceElement + " not specified"); } String[] dfsNameNodeElements = dfsNameNodes.split(","); for (String dfsNameNodeElement : dfsNameNodeElements) { String nameNodeUrlKey = "dfs.namenode.rpc-address." + dfsNameserviceElement + "." + dfsNameNodeElement.trim(); String nameNodeUrl = configs.get(nameNodeUrlKey); nameNodeUrl = (nameNodeUrl == null) ? "" : nameNodeUrl.trim(); if (nameNodeUrl.isEmpty()) { throw new IllegalArgumentException("Value for " + nameNodeUrlKey + " not specified"); } } } } } HdfsClient(String serviceName, Map<String,String> connectionProperties); List<String> listFiles(final String baseDir, final String fileMatching, final List<String> pathList); static final void main(String[] args); static Map<String, Object> connectionTest(String serviceName,
			Map<String, String> configs); static void validateConnectionConfigs(Map<String, String> configs);  }
@Test(expected = IllegalArgumentException.class) public void testUsernameNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testPasswordNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testAuthenticationNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testFsDefaultNameNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testProxyProviderNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: configs.put("dfs.nameservices", "hwqe-1425428405"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testNnElementsNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: configs.put("dfs.nameservices", "hwqe-1425428405"); configs.put("dfs.client.failover.proxy.provider.hwqe-1425428405", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testNn1UrlNn2UrlNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: configs.put("dfs.nameservices", "hwqe-1425428405"); configs.put("dfs.client.failover.proxy.provider.hwqe-1425428405", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); configs.put("dfs.ha.namenodes.hwqe-1425428405", "nn1,nn2"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testNn1UrlNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: configs.put("dfs.nameservices", "hwqe-1425428405"); configs.put("dfs.client.failover.proxy.provider.hwqe-1425428405", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); configs.put("dfs.ha.namenodes.hwqe-1425428405", "nn1,nn2"); configs.put("dfs.namenode.rpc-address.hwqe-1425428405.nn2", "node-2.example.com:8020"); HdfsClient.validateConnectionConfigs(configs); }
@Test(expected = IllegalArgumentException.class) public void testNn2UrlNotSpecified() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: configs.put("dfs.nameservices", "hwqe-1425428405"); configs.put("dfs.client.failover.proxy.provider.hwqe-1425428405", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); configs.put("dfs.ha.namenodes.hwqe-1425428405", "nn1,nn2"); configs.put("dfs.namenode.rpc-address.hwqe-1425428405.nn1", "node-1.example.com:8020"); HdfsClient.validateConnectionConfigs(configs); }
@Test public void testValidNonHaConfig() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: HdfsClient.validateConnectionConfigs(configs); }
@Test public void testValidHaConfig() throws IllegalArgumentException { Map<String, String> configs = new HashMap<String, String>(); configs.put("username", "hdfsuser"); configs.put("password", "hdfsuser"); configs.put("hadoop.security.authentication", "simple"); configs.put("fs.default.name", "hdfs: configs.put("fs.default.name", "hdfs: configs.put("dfs.nameservices", "hwqe-1425428405"); configs.put("dfs.client.failover.proxy.provider.hwqe-1425428405", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); configs.put("dfs.ha.namenodes.hwqe-1425428405", "nn1,nn2"); configs.put("dfs.namenode.rpc-address.hwqe-1425428405.nn1", "node-1.example.com:8020"); configs.put("dfs.namenode.rpc-address.hwqe-1425428405.nn2", "node-2.example.com:8020"); HdfsClient.validateConnectionConfigs(configs); }