private static NamenodeProtocol createNNProxyWithNamenodeProtocol( InetSocketAddress address, Configuration conf, UserGroupInformation ugi, boolean withRetries) throws IOException { NamenodeProtocolPB proxy = (NamenodeProtocolPB) createNameNodeProxy( address, conf, ugi, NamenodeProtocolPB.class); if (withRetries) { // create the proxy with retries RetryPolicy timeoutPolicy = RetryPolicies.exponentialBackoffRetry(5, 200, TimeUnit.MILLISECONDS); Map<String, RetryPolicy> methodNameToPolicyMap = new HashMap<String, RetryPolicy>(); methodNameToPolicyMap.put("getBlocks", timeoutPolicy); methodNameToPolicyMap.put("getAccessKeys", timeoutPolicy); NamenodeProtocol translatorProxy = new NamenodeProtocolTranslatorPB(proxy); return (NamenodeProtocol) RetryProxy.create( NamenodeProtocol.class, translatorProxy, methodNameToPolicyMap); } else { return new NamenodeProtocolTranslatorPB(proxy); } }
@Test public void testNamenodeProtocol() throws IOException { NamenodeProtocol np = NameNodeProxies.createNonHAProxy(conf, nnAddress, NamenodeProtocol.class, UserGroupInformation.getCurrentUser(), true).getProxy(); boolean exists = RpcClientUtil.isMethodSupported(np, NamenodeProtocolPB.class, RPC.RpcKind.RPC_PROTOCOL_BUFFER, RPC.getProtocolVersion(NamenodeProtocolPB.class), "rollEditLog"); assertTrue(exists); exists = RpcClientUtil.isMethodSupported(np, NamenodeProtocolPB.class, RPC.RpcKind.RPC_PROTOCOL_BUFFER, RPC.getProtocolVersion(NamenodeProtocolPB.class), "bogusMethod"); assertFalse(exists); }
private static NamenodeProtocol createNNProxyWithNamenodeProtocol( InetSocketAddress address, Configuration conf, UserGroupInformation ugi, boolean withRetries) throws IOException { NamenodeProtocolPB proxy = (NamenodeProtocolPB) createNameNodeProxy( address, conf, ugi, NamenodeProtocolPB.class, 0); if (withRetries) { // create the proxy with retries RetryPolicy timeoutPolicy = RetryPolicies.exponentialBackoffRetry(5, 200, TimeUnit.MILLISECONDS); Map<String, RetryPolicy> methodNameToPolicyMap = new HashMap<String, RetryPolicy>(); methodNameToPolicyMap.put("getBlocks", timeoutPolicy); methodNameToPolicyMap.put("getAccessKeys", timeoutPolicy); NamenodeProtocol translatorProxy = new NamenodeProtocolTranslatorPB(proxy); return (NamenodeProtocol) RetryProxy.create( NamenodeProtocol.class, translatorProxy, methodNameToPolicyMap); } else { return new NamenodeProtocolTranslatorPB(proxy); } }
public void start() throws IOException { int rpcHandlerCount = conf.getInt(ProxyConfig.PROXY_HANDLER_COUNT, ProxyConfig.PROXY_HANDLER_COUNT_DEFAULT); RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class, ProtobufRpcEngine.class); RPC.setProtocolEngine(conf, NamenodeProtocolPB.class, ProtobufRpcEngine.class); this.protocol = (ClientProtocol) Proxy.newProxyInstance( this.getClass().getClassLoader(), new Class[]{ClientProtocol.class}, this.invocationHandler); ClientNamenodeProtocolPB proxy = new ClientNamenodeProtocolServerSideTranslatorPB(this.protocol); BlockingService clientNNPbService = ClientNamenodeProtocolProtos.ClientNamenodeProtocol. newReflectiveBlockingService(proxy); int port = conf.getInt(ProxyConfig.RPC_PORT, ProxyConfig.RPC_PORT_DEFAULT); this.rpcServer = new RPC.Builder(conf) .setProtocol(org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolPB.class) .setInstance(clientNNPbService).setBindAddress("0.0.0.0") .setPort(port).setNumHandlers(rpcHandlerCount) .setVerbose(false).build(); this.rpcServer.start(); InetSocketAddress listenAddr = rpcServer.getListenerAddress(); rpcAddress = new InetSocketAddress("0.0.0.0", listenAddr.getPort()); }