From 8affaa6312300a89ed1eb12bcae3374f15a30e19 Mon Sep 17 00:00:00 2001 From: Symious Date: Thu, 9 Sep 2021 09:57:27 +0800 Subject: [PATCH] HDFS-16210. RBF: Add the option of refreshCallQueue to RouterAdmin (#3379) (cherry picked from commit c0890e6d04dda6f2716d07427816721fbdf9c3b4) --- .../federation/router/RouterAdminServer.java | 22 ++++++++++- .../hdfs/tools/federation/RouterAdmin.java | 37 +++++++++++++++++++ .../federation/router/TestRouterAdminCLI.java | 9 +++++ 3 files changed, 67 insertions(+), 1 deletion(-) diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java index 159b1033e58..4dd0693e97a 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterAdminServer.java @@ -78,11 +78,15 @@ import org.apache.hadoop.hdfs.server.namenode.NameNode; import org.apache.hadoop.ipc.ProtobufRpcEngine2; import org.apache.hadoop.ipc.RPC; import org.apache.hadoop.ipc.RPC.Server; +import org.apache.hadoop.ipc.RefreshCallQueueProtocol; import org.apache.hadoop.ipc.RefreshRegistry; import org.apache.hadoop.ipc.RefreshResponse; import org.apache.hadoop.ipc.proto.GenericRefreshProtocolProtos; +import org.apache.hadoop.ipc.proto.RefreshCallQueueProtocolProtos; import org.apache.hadoop.ipc.protocolPB.GenericRefreshProtocolPB; import org.apache.hadoop.ipc.protocolPB.GenericRefreshProtocolServerSideTranslatorPB; +import org.apache.hadoop.ipc.protocolPB.RefreshCallQueueProtocolPB; +import org.apache.hadoop.ipc.protocolPB.RefreshCallQueueProtocolServerSideTranslatorPB; import org.apache.hadoop.security.AccessControlException; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.security.authorize.ProxyUsers; @@ -98,7 +102,7 @@ import org.apache.hadoop.thirdparty.protobuf.BlockingService; * router. It is created, started, and stopped by {@link Router}. */ public class RouterAdminServer extends AbstractService - implements RouterAdminProtocol { + implements RouterAdminProtocol, RefreshCallQueueProtocol { private static final Logger LOG = LoggerFactory.getLogger(RouterAdminServer.class); @@ -184,8 +188,16 @@ public class RouterAdminServer extends AbstractService GenericRefreshProtocolProtos.GenericRefreshProtocolService. newReflectiveBlockingService(genericRefreshXlator); + RefreshCallQueueProtocolServerSideTranslatorPB refreshCallQueueXlator = + new RefreshCallQueueProtocolServerSideTranslatorPB(this); + BlockingService refreshCallQueueService = + RefreshCallQueueProtocolProtos.RefreshCallQueueProtocolService. + newReflectiveBlockingService(refreshCallQueueXlator); + DFSUtil.addPBProtocol(conf, GenericRefreshProtocolPB.class, genericRefreshService, adminServer); + DFSUtil.addPBProtocol(conf, RefreshCallQueueProtocolPB.class, + refreshCallQueueService, adminServer); } /** @@ -643,4 +655,12 @@ public class RouterAdminServer extends AbstractService ProxyUsers.refreshSuperUserGroupsConfiguration(); return true; } + + @Override // RefreshCallQueueProtocol + public void refreshCallQueue() throws IOException { + LOG.info("Refreshing call queue."); + + Configuration configuration = new Configuration(); + router.getRpcServer().getServer().refreshCallQueue(configuration); + } } diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java index 7422989d6aa..deadf3d3132 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java @@ -77,6 +77,8 @@ import org.apache.hadoop.ipc.RefreshResponse; import org.apache.hadoop.ipc.RemoteException; import org.apache.hadoop.ipc.protocolPB.GenericRefreshProtocolClientSideTranslatorPB; import org.apache.hadoop.ipc.protocolPB.GenericRefreshProtocolPB; +import org.apache.hadoop.ipc.protocolPB.RefreshCallQueueProtocolClientSideTranslatorPB; +import org.apache.hadoop.ipc.protocolPB.RefreshCallQueueProtocolPB; import org.apache.hadoop.net.NetUtils; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.util.StringUtils; @@ -388,6 +390,8 @@ public class RouterAdmin extends Configured implements Tool { exitCode = genericRefresh(argv, i); } else if ("-refreshSuperUserGroupsConfiguration".equals(cmd)) { exitCode = refreshSuperUserGroupsConfiguration(); + } else if ("-refreshCallQueue".equals(cmd)) { + exitCode = refreshCallQueue(); } else { throw new IllegalArgumentException("Unknown Command: " + cmd); } @@ -1258,6 +1262,39 @@ public class RouterAdmin extends Configured implements Tool { } } + /** + * Refresh Router's call Queue. + * + * @throws IOException if the operation was not successful. + */ + private int refreshCallQueue() throws IOException { + Configuration conf = getConf(); + String hostport = getConf().getTrimmed( + RBFConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_KEY, + RBFConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_DEFAULT); + + // Create the client + Class xface = RefreshCallQueueProtocolPB.class; + InetSocketAddress address = NetUtils.createSocketAddr(hostport); + UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); + + RPC.setProtocolEngine(conf, xface, ProtobufRpcEngine2.class); + RefreshCallQueueProtocolPB proxy = (RefreshCallQueueProtocolPB)RPC.getProxy( + xface, RPC.getProtocolVersion(xface), address, ugi, conf, + NetUtils.getDefaultSocketFactory(conf), 0); + + int returnCode = -1; + try (RefreshCallQueueProtocolClientSideTranslatorPB xlator = + new RefreshCallQueueProtocolClientSideTranslatorPB(proxy)) { + xlator.refreshCallQueue(); + System.out.println("Refresh call queue successfully for " + hostport); + returnCode = 0; + } catch (IOException ioe){ + System.out.println("Refresh call queue unsuccessfully for " + hostport); + } + return returnCode; + } + /** * Normalize a path for that filesystem. * diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java index 837607cea52..14a596546f8 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java @@ -1559,6 +1559,15 @@ public class TestRouterAdminCLI { assertEquals(0, ToolRunner.run(admin, argv)); } + @Test + public void testRefreshCallQueue() throws Exception { + + System.setOut(new PrintStream(out)); + String[] argv = new String[]{"-refreshCallQueue"}; + assertEquals(0, ToolRunner.run(admin, argv)); + assertTrue(out.toString().contains("Refresh call queue successfully")); + } + private void addMountTable(String src, String nsId, String dst) throws Exception { String[] argv = new String[] {"-add", src, nsId, dst};