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 b44142fb17..d2b20bc4e8 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 @@ -81,11 +81,15 @@ 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; @@ -102,7 +106,7 @@ * 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); @@ -197,8 +201,16 @@ public RouterAdminServer(Configuration conf, Router router) 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); } /** @@ -764,4 +776,12 @@ public boolean refreshSuperUserGroupsConfiguration() throws IOException { 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 7422989d6a..deadf3d313 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.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 int run(String[] argv) throws Exception { 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 int genericRefresh(String[] argv, int i) throws IOException { } } + /** + * 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 1daff053ed..4134b49d9a 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 @@ -1740,6 +1740,15 @@ public void testErrorFaultTolerant() throws Exception { 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};