Repository: hadoop Updated Branches: refs/heads/branch-2.9 42c47971d -> 9c8a79806
HDFS-13750. RBF: Router ID in RouterRpcClient is always null. Contributed by Takanobu Asanuma. (cherry picked from commit 18ebe183052ee37ce041432f3b32a3b9a531c3e6) Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/9c8a7980 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/9c8a7980 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/9c8a7980 Branch: refs/heads/branch-2.9 Commit: 9c8a79806d2218c6dfd79d373030074ea981d0c4 Parents: 42c4797 Author: Yiqun Lin <[email protected]> Authored: Mon Aug 20 23:08:26 2018 +0800 Committer: Yiqun Lin <[email protected]> Committed: Mon Aug 20 23:10:02 2018 +0800 ---------------------------------------------------------------------- .../federation/router/RouterRpcClient.java | 18 +++++++++------- .../federation/router/RouterRpcServer.java | 3 ++- .../hdfs/server/federation/MockResolver.java | 3 +++ .../server/federation/router/TestRouter.java | 22 ++++++++++++++++++++ 4 files changed, 37 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/9c8a7980/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java index e4d304d..0161900 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java @@ -92,8 +92,8 @@ public class RouterRpcClient { LoggerFactory.getLogger(RouterRpcClient.class); - /** Router identifier. */ - private final String routerId; + /** Router using this RPC client. */ + private final Router router; /** Interface to identify the active NN for a nameservice or blockpool ID. */ private final ActiveNamenodeResolver namenodeResolver; @@ -116,12 +116,13 @@ public class RouterRpcClient { * Create a router RPC client to manage remote procedure calls to NNs. * * @param conf Hdfs Configuation. + * @param router A router using this RPC client. * @param resolver A NN resolver to determine the currently active NN in HA. * @param monitor Optional performance monitor. */ - public RouterRpcClient(Configuration conf, String identifier, + public RouterRpcClient(Configuration conf, Router router, ActiveNamenodeResolver resolver, RouterRpcMonitor monitor) { - this.routerId = identifier; + this.router = router; this.namenodeResolver = resolver; @@ -335,7 +336,8 @@ public class RouterRpcClient { if (namenodes == null || namenodes.isEmpty()) { throw new IOException("No namenodes to invoke " + method.getName() + - " with params " + Arrays.toString(params) + " from " + this.routerId); + " with params " + Arrays.toString(params) + " from " + + router.getRouterId()); } Object ret = null; @@ -1118,7 +1120,7 @@ public class RouterRpcClient { String msg = "Not enough client threads " + active + "/" + total; LOG.error(msg); throw new StandbyException( - "Router " + routerId + " is overloaded: " + msg); + "Router " + router.getRouterId() + " is overloaded: " + msg); } catch (InterruptedException ex) { LOG.error("Unexpected error while invoking API: {}", ex.getMessage()); throw new IOException( @@ -1142,7 +1144,7 @@ public class RouterRpcClient { if (namenodes == null || namenodes.isEmpty()) { throw new IOException("Cannot locate a registered namenode for " + nsId + - " from " + this.routerId); + " from " + router.getRouterId()); } return namenodes; } @@ -1163,7 +1165,7 @@ public class RouterRpcClient { if (namenodes == null || namenodes.isEmpty()) { throw new IOException("Cannot locate a registered namenode for " + bpId + - " from " + this.routerId); + " from " + router.getRouterId()); } return namenodes; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/9c8a7980/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java index 65e27d3..3c753e1 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java @@ -193,6 +193,7 @@ public class RouterRpcServer extends AbstractService * Construct a router RPC server. * * @param configuration HDFS Configuration. + * @param router A router using this RPC server. * @param nnResolver The NN resolver instance to determine active NNs in HA. * @param fileResolver File resolver to resolve file paths to subclusters. * @throws IOException If the RPC server could not be created. @@ -294,7 +295,7 @@ public class RouterRpcServer extends AbstractService this.rpcMonitor = ReflectionUtils.newInstance(rpcMonitorClass, conf); // Create the client - this.rpcClient = new RouterRpcClient(this.conf, this.router.getRouterId(), + this.rpcClient = new RouterRpcClient(this.conf, this.router, this.namenodeResolver, this.rpcMonitor); // Initialize modules http://git-wip-us.apache.org/repos/asf/hadoop/blob/9c8a7980/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java index 36cce39..f5636ce 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java @@ -129,6 +129,9 @@ public class MockResolver // Return a copy of the list because it is updated periodically List<? extends FederationNamenodeContext> namenodes = this.resolver.get(nameserviceId); + if (namenodes == null) { + namenodes = new ArrayList<>(); + } return Collections.unmodifiableList(new ArrayList<>(namenodes)); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/9c8a7980/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouter.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouter.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouter.java index f8cf009..75b67e2 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouter.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouter.java @@ -20,7 +20,9 @@ package org.apache.hadoop.hdfs.server.federation.router; import static org.apache.hadoop.test.GenericTestUtils.assertExceptionContains; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import java.io.IOException; import java.net.InetSocketAddress; @@ -185,4 +187,24 @@ public class TestRouter { router.stop(); router.close(); } + + @Test + public void testRouterIDInRouterRpcClient() throws Exception { + + Router router = new Router(); + router.init(new RouterConfigBuilder(conf).rpc().build()); + router.setRouterId("Router-0"); + RemoteMethod remoteMethod = mock(RemoteMethod.class); + + try { + router.getRpcServer().getRPCClient() + .invokeSingle("ns0", remoteMethod); + fail("Exception should have been thrown"); + } catch (IOException e) { + assertTrue(e.getMessage().contains("Router-0")); + } + + router.stop(); + router.close(); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
