RPC { public static <T> T getProxy(Class<T> protocol, long clientVersion, InetSocketAddress addr, Configuration conf, SocketFactory factory) throws IOException { return getProtocolProxy( protocol, clientVersion, addr, conf, factory).getProxy(); } private  RPC(); static String getProtocolName(Class<?> protocol); static long getProtocolVersion(Class<?> protocol); static void setProtocolEngine(Configuration conf,
                                Class<?> protocol, Class<?> engine); static T waitForProxy(
      Class<T> protocol,
      long clientVersion,
      InetSocketAddress addr,
      Configuration conf
      ); static ProtocolProxy<T> waitForProtocolProxy(Class<T> protocol,
                             long clientVersion,
                             InetSocketAddress addr,
                             Configuration conf); static T waitForProxy(Class<T> protocol, long clientVersion,
                             InetSocketAddress addr, Configuration conf,
                             long connTimeout); static ProtocolProxy<T> waitForProtocolProxy(Class<T> protocol,
                             long clientVersion,
                             InetSocketAddress addr, Configuration conf,
                             long connTimeout); static T waitForProxy(Class<T> protocol,
                             long clientVersion,
                             InetSocketAddress addr, Configuration conf,
                             int rpcTimeout,
                             long timeout); static ProtocolProxy<T> waitForProtocolProxy(Class<T> protocol,
                               long clientVersion,
                               InetSocketAddress addr, Configuration conf,
                               int rpcTimeout,
                               RetryPolicy connectionRetryPolicy,
                               long timeout); static T getProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr, Configuration conf,
                                SocketFactory factory); static ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr, Configuration conf,
                                SocketFactory factory); static T getProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr,
                                UserGroupInformation ticket,
                                Configuration conf,
                                SocketFactory factory); static ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr,
                                UserGroupInformation ticket,
                                Configuration conf,
                                SocketFactory factory); static T getProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr,
                                UserGroupInformation ticket,
                                Configuration conf,
                                SocketFactory factory,
                                int rpcTimeout); static ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr,
                                UserGroupInformation ticket,
                                Configuration conf,
                                SocketFactory factory,
                                int rpcTimeout,
                                RetryPolicy connectionRetryPolicy); static ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr,
                                UserGroupInformation ticket,
                                Configuration conf,
                                SocketFactory factory,
                                int rpcTimeout,
                                RetryPolicy connectionRetryPolicy,
                                AtomicBoolean fallbackToSimpleAuth); static T getProxy(Class<T> protocol,
                                 long clientVersion,
                                 InetSocketAddress addr, Configuration conf); static InetSocketAddress getServerAddress(Object proxy); static ConnectionId getConnectionIdForProxy(Object proxy); static ProtocolProxy<T> getProtocolProxy(Class<T> protocol,
                                long clientVersion,
                                InetSocketAddress addr, Configuration conf); static void stopProxy(Object proxy);  }
@Test(timeout=90000) public void testRPCInterruptedSimple() throws IOException { final Configuration conf = new Configuration(); Server server = new RPC.Builder(conf).setProtocol(TestProtocol.class) .setInstance(new TestImpl()).setBindAddress(ADDRESS) .setPort(0).setNumHandlers(5).setVerbose(true) .setSecretManager(null).build(); server.start(); try { InetSocketAddress addr = NetUtils.getConnectAddress(server); final TestProtocol proxy = RPC.getProxy( TestProtocol.class, TestProtocol.versionID, addr, conf); proxy.ping(); Thread.currentThread().interrupt(); try { proxy.ping(); fail("Interruption did not cause IPC to fail"); } catch (IOException ioe) { if (ioe.toString().contains("InterruptedException") || ioe instanceof InterruptedIOException) { Thread.interrupted(); return; } throw ioe; } } finally { server.stop(); } }
@Test(timeout=30000) public void testRPCInterrupted() throws IOException, InterruptedException { final Configuration conf = new Configuration(); Server server = new RPC.Builder(conf).setProtocol(TestProtocol.class) .setInstance(new TestImpl()).setBindAddress(ADDRESS) .setPort(0).setNumHandlers(5).setVerbose(true) .setSecretManager(null).build(); server.start(); try { int numConcurrentRPC = 200; InetSocketAddress addr = NetUtils.getConnectAddress(server); final CyclicBarrier barrier = new CyclicBarrier(numConcurrentRPC); final CountDownLatch latch = new CountDownLatch(numConcurrentRPC); final AtomicBoolean leaderRunning = new AtomicBoolean(true); final AtomicReference<Throwable> error = new AtomicReference<Throwable>(); Thread leaderThread = null; for (int i = 0; i < numConcurrentRPC; i++) { final int num = i; final TestProtocol proxy = RPC.getProxy(TestProtocol.class, TestProtocol.versionID, addr, conf); Thread rpcThread = new Thread(new Runnable() { @Override public void run() { try { barrier.await(); while (num == 0 || leaderRunning.get()) { proxy.slowPing(false); } proxy.slowPing(false); } catch (Exception e) { if (num == 0) { leaderRunning.set(false); } else { error.set(e); } LOG.error("thread " + num, e); } finally { latch.countDown(); } } }); rpcThread.start(); if (leaderThread == null) { leaderThread = rpcThread; } } Thread.sleep(1000); while (leaderRunning.get()) { leaderThread.interrupt(); } latch.await(); assertTrue("rpc got exception " + error.get(), error.get() == null); } finally { server.stop(); } }
@Test(timeout=90000) public void testRPCInterruptedSimple() throws IOException { final Configuration conf = new Configuration(); Server server = new RPC.Builder(conf).setProtocol(TestProtocol.class) .setInstance(new TestImpl()).setBindAddress(ADDRESS) .setPort(0).setNumHandlers(5).setVerbose(true) .setSecretManager(null).build(); server.start(); InetSocketAddress addr = NetUtils.getConnectAddress(server); final TestProtocol proxy = RPC.getProxy( TestProtocol.class, TestProtocol.versionID, addr, conf); proxy.ping(); Thread.currentThread().interrupt(); try { proxy.ping(); fail("Interruption did not cause IPC to fail"); } catch (IOException ioe) { if (!ioe.toString().contains("InterruptedException")) { throw ioe; } Thread.interrupted(); } finally { server.stop(); } }
@Test(timeout=30000) public void testRPCInterrupted() throws IOException, InterruptedException { final Configuration conf = new Configuration(); Server server = new RPC.Builder(conf).setProtocol(TestProtocol.class) .setInstance(new TestImpl()).setBindAddress(ADDRESS) .setPort(0).setNumHandlers(5).setVerbose(true) .setSecretManager(null).build(); server.start(); int numConcurrentRPC = 200; InetSocketAddress addr = NetUtils.getConnectAddress(server); final CyclicBarrier barrier = new CyclicBarrier(numConcurrentRPC); final CountDownLatch latch = new CountDownLatch(numConcurrentRPC); final AtomicBoolean leaderRunning = new AtomicBoolean(true); final AtomicReference<Throwable> error = new AtomicReference<Throwable>(); Thread leaderThread = null; for (int i = 0; i < numConcurrentRPC; i++) { final int num = i; final TestProtocol proxy = RPC.getProxy( TestProtocol.class, TestProtocol.versionID, addr, conf); Thread rpcThread = new Thread(new Runnable() { @Override public void run() { try { barrier.await(); while (num == 0 || leaderRunning.get()) { proxy.slowPing(false); } proxy.slowPing(false); } catch (Exception e) { if (num == 0) { leaderRunning.set(false); } else { error.set(e); } LOG.error(e); } finally { latch.countDown(); } } }); rpcThread.start(); if (leaderThread == null) { leaderThread = rpcThread; } } Thread.sleep(1000); while (leaderRunning.get()) { leaderThread.interrupt(); } latch.await(); assertTrue("rpc got exception " + error.get(), error.get() == null); server.stop(); }