@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;
    }
  }
  // let threads get past the barrier
  Thread.sleep(1000);
  // stop a single thread
  while (leaderRunning.get()) {
    leaderThread.interrupt();
  }
  
  latch.await();
  
  // should not cause any other thread to get an error
  assertTrue("rpc got exception " + error.get(), error.get() == null);
  server.stop();
}