/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.ipc;

import com.google.common.base.Supplier;
import com.google.protobuf.ServiceException;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.io.retry.RetryProxy;
import org.apache.hadoop.ipc.Client.ConnectionId;
import org.apache.hadoop.ipc.Server.Call;
import org.apache.hadoop.ipc.Server.Connection;
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcErrorCodeProto;
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto;
import org.apache.hadoop.ipc.protobuf.TestProtos;
import org.apache.hadoop.metrics2.MetricsRecordBuilder;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.hadoop.security.authorize.AuthorizationException;
import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.security.authorize.Service;
import org.apache.hadoop.security.token.SecretManager;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.test.MetricsAsserts;
import org.apache.hadoop.test.MockitoUtil;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.internal.util.reflection.Whitebox;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;

import javax.net.SocketFactory;
import java.io.Closeable;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static org.apache.hadoop.test.MetricsAsserts.assertCounter;
import static org.apache.hadoop.test.MetricsAsserts.assertCounterGt;
import static org.apache.hadoop.test.MetricsAsserts.assertGauge;
import static org.apache.hadoop.test.MetricsAsserts.getDoubleGauge;
import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;

/** Unit tests for RPC. */
@SuppressWarnings("deprecation")
public class TestRPC extends TestRpcBase {

  public static final Logger LOG = LoggerFactory.getLogger(TestRPC.class);

  @Before
  public void setup() {
    setupConf();
  }

  int datasize = 1024*100;
  int numThreads = 50;

  public interface TestProtocol extends VersionedProtocol {
    long versionID = 1L;

    void ping() throws IOException;
    void sleep(long delay) throws IOException, InterruptedException;
    String echo(String value) throws IOException;
    String[] echo(String[] value) throws IOException;
    Writable echo(Writable value) throws IOException;
    int add(int v1, int v2) throws IOException;
    int add(int[] values) throws IOException;
    int error() throws IOException;
  }

  public static class TestImpl implements TestProtocol {
    int fastPingCounter = 0;
    
    @Override
    public long getProtocolVersion(String protocol, long clientVersion) {
      return TestProtocol.versionID;
    }

    @Override
    public ProtocolSignature getProtocolSignature(String protocol, long clientVersion,
                                                  int hashcode) {
      return new ProtocolSignature(TestProtocol.versionID, null);
    }

    @Override
    public void ping() {}
    
    @Override
    public void sleep(long delay) throws InterruptedException {
      Thread.sleep(delay);
    }

    @Override
    public String echo(String value) throws IOException { return value; }

    @Override
    public String[] echo(String[] values) throws IOException { return values; }

    @Override
    public Writable echo(Writable writable) {
      return writable;
    }
    @Override
    public int add(int v1, int v2) {
      return v1 + v2;
    }

    @Override
    public int add(int[] values) {
      int sum = 0;
      for (int i = 0; i < values.length; i++) {
        sum += values[i];
      }
      return sum;
    }

    @Override
    public int error() throws IOException {
      throw new IOException("bobo");
    }
  }

  //
  // an object that does a bunch of transactions
  //
  static class Transactions implements Runnable {
    int datasize;
    TestRpcService proxy;

    Transactions(TestRpcService proxy, int datasize) {
      this.proxy = proxy;
      this.datasize = datasize;
    }

    // do two RPC that transfers data.
    @Override
    public void run() {
      Integer[] indata = new Integer[datasize];
      Arrays.fill(indata, 123);
      TestProtos.ExchangeRequestProto exchangeRequest =
          TestProtos.ExchangeRequestProto.newBuilder().addAllValues(
              Arrays.asList(indata)).build();
      Integer[] outdata = null;
      TestProtos.ExchangeResponseProto exchangeResponse;

      TestProtos.AddRequestProto addRequest =
          TestProtos.AddRequestProto.newBuilder().setParam1(1)
              .setParam2(2).build();
      TestProtos.AddResponseProto addResponse;

      int val = 0;
      try {
        exchangeResponse = proxy.exchange(null, exchangeRequest);
        outdata = new Integer[exchangeResponse.getValuesCount()];
        outdata = exchangeResponse.getValuesList().toArray(outdata);
        addResponse = proxy.add(null, addRequest);
        val = addResponse.getResult();
      } catch (ServiceException e) {
        assertTrue("Exception from RPC exchange() "  + e, false);
      }
      assertEquals(indata.length, outdata.length);
      assertEquals(3, val);
      for (int i = 0; i < outdata.length; i++) {
        assertEquals(outdata[i].intValue(), i);
      }
    }
  }

  //
  // A class that does an RPC but does not read its response.
  //
  static class SlowRPC implements Runnable {
    private TestRpcService proxy;
    private volatile boolean done;

    SlowRPC(TestRpcService proxy) {
      this.proxy = proxy;
      done = false;
    }

    boolean isDone() {
      return done;
    }

    @Override
    public void run() {
      try {
        // this would hang until two fast pings happened
        ping(true);
        done = true;
      } catch (ServiceException e) {
        assertTrue("SlowRPC ping exception " + e, false);
      }
    }

    void ping(boolean shouldSlow) throws ServiceException {
      // this would hang until two fast pings happened
      proxy.slowPing(null, newSlowPingRequest(shouldSlow));
    }
  }

  /**
   * A basic interface for testing client-side RPC resource cleanup.
   */
  private interface StoppedProtocol {
    long versionID = 0;

    void stop();
  }

  /**
   * A class used for testing cleanup of client side RPC resources.
   */
  private static class StoppedRpcEngine implements RpcEngine {

    @Override
    public <T> ProtocolProxy<T> getProxy(
        Class<T> protocol, long clientVersion, InetSocketAddress addr,
        UserGroupInformation ticket, Configuration conf,
        SocketFactory factory, int rpcTimeout,
        RetryPolicy connectionRetryPolicy) throws IOException {
      return getProxy(protocol, clientVersion, addr, ticket, conf, factory,
          rpcTimeout, connectionRetryPolicy, null, null);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> ProtocolProxy<T> getProxy(
        Class<T> protocol, long clientVersion, InetSocketAddress addr,
        UserGroupInformation ticket, Configuration conf, SocketFactory factory,
        int rpcTimeout, RetryPolicy connectionRetryPolicy,
        AtomicBoolean fallbackToSimpleAuth, AlignmentContext alignmentContext)
        throws IOException {
      T proxy = (T) Proxy.newProxyInstance(protocol.getClassLoader(),
          new Class[] { protocol }, new StoppedInvocationHandler());
      return new ProtocolProxy<T>(protocol, proxy, false);
    }

    @Override
    public org.apache.hadoop.ipc.RPC.Server getServer(
        Class<?> protocol, Object instance, String bindAddress, int port,
        int numHandlers, int numReaders, int queueSizePerHandler,
        boolean verbose, Configuration conf,
        SecretManager<? extends TokenIdentifier> secretManager,
        String portRangeConfig, AlignmentContext alignmentContext)
        throws IOException {
      return null;
    }

    @Override
    public ProtocolProxy<ProtocolMetaInfoPB> getProtocolMetaInfoProxy(
        ConnectionId connId, Configuration conf, SocketFactory factory)
        throws IOException {
      throw new UnsupportedOperationException("This proxy is not supported");
    }
  }

  /**
   * An invocation handler which does nothing when invoking methods, and just
   * counts the number of times close() is called.
   */
  private static class StoppedInvocationHandler
      implements InvocationHandler, Closeable {

    private int closeCalled = 0;

    @Override
    public Object invoke(Object proxy, Method method, Object[] args)
        throws Throwable {
      return null;
    }

    @Override
    public void close() throws IOException {
      closeCalled++;
    }

    public int getCloseCalled() {
      return closeCalled;
    }

  }

  @Test
  public void testConfRpc() throws IOException {
    Server server = newServerBuilder(conf)
        .setNumHandlers(1).setVerbose(false).build();

    // Just one handler
    int confQ = conf.getInt(
        CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_KEY,
        CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT);
    assertEquals(confQ, server.getMaxQueueSize());

    int confReaders = conf.getInt(
        CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
        CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);
    assertEquals(confReaders, server.getNumReaders());

    server = newServerBuilder(conf)
        .setNumHandlers(1).setnumReaders(3).setQueueSizePerHandler(200)
        .setVerbose(false).build();

    assertEquals(3, server.getNumReaders());
    assertEquals(200, server.getMaxQueueSize());

    server = newServerBuilder(conf).setQueueSizePerHandler(10)
        .setNumHandlers(2).setVerbose(false).build();
    assertEquals(2 * 10, server.getMaxQueueSize());
  }

  @Test
  public void testProxyAddress() throws Exception {
    Server server = null;
    TestRpcService proxy = null;

    try {
      server = setupTestServer(conf, -1);
      // create a client
      proxy = getClient(addr, conf);
      assertEquals(addr, RPC.getServerAddress(proxy));
    } finally {
      stop(server, proxy);
    }
  }

  @Test
  public void testSlowRpc() throws IOException, ServiceException {
    Server server;
    TestRpcService proxy = null;

    System.out.println("Testing Slow RPC");
    // create a server with two handlers
    server = setupTestServer(conf, 2);

    try {
      // create a client
      proxy = getClient(addr, conf);

      SlowRPC slowrpc = new SlowRPC(proxy);
      Thread thread = new Thread(slowrpc, "SlowRPC");
      thread.start(); // send a slow RPC, which won't return until two fast pings
      assertTrue("Slow RPC should not have finished1.", !slowrpc.isDone());

      slowrpc.ping(false); // first fast ping

      // verify that the first RPC is still stuck
      assertTrue("Slow RPC should not have finished2.", !slowrpc.isDone());

      slowrpc.ping(false); // second fast ping

      // Now the slow ping should be able to be executed
      while (!slowrpc.isDone()) {
        System.out.println("Waiting for slow RPC to get done.");
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {}
      }
    } finally {
      System.out.println("Down slow rpc testing");
      stop(server, proxy);
    }
  }

  @Test
  public void testCalls() throws Exception {
    testCallsInternal(conf);
  }

  private void testCallsInternal(Configuration myConf) throws Exception {
    Server server;
    TestRpcService proxy = null;

    server = setupTestServer(myConf, -1);
    try {
      proxy = getClient(addr, myConf);

      proxy.ping(null, newEmptyRequest());

      TestProtos.EchoResponseProto echoResp = proxy.echo(null,
          newEchoRequest("foo"));
      assertEquals(echoResp.getMessage(), "foo");

      echoResp = proxy.echo(null, newEchoRequest(""));
      assertEquals(echoResp.getMessage(), "");

      // Check rpcMetrics
      MetricsRecordBuilder rb = getMetrics(server.rpcMetrics.name());
      assertCounter("RpcProcessingTimeNumOps", 3L, rb);
      assertCounterGt("SentBytes", 0L, rb);
      assertCounterGt("ReceivedBytes", 0L, rb);

      // Number of calls to echo method should be 2
      rb = getMetrics(server.rpcDetailedMetrics.name());
      assertCounter("EchoNumOps", 2L, rb);

      // Number of calls to ping method should be 1
      assertCounter("PingNumOps", 1L, rb);

      String[] strings = new String[] {"foo","bar"};
      TestProtos.EchoRequestProto2 echoRequest2 =
          TestProtos.EchoRequestProto2.newBuilder().addAllMessage(
              Arrays.asList(strings)).build();
      TestProtos.EchoResponseProto2 echoResponse2 =
          proxy.echo2(null, echoRequest2);
      assertTrue(Arrays.equals(echoResponse2.getMessageList().toArray(),
          strings));

      echoRequest2 = TestProtos.EchoRequestProto2.newBuilder()
          .addAllMessage(Collections.<String>emptyList()).build();
      echoResponse2 = proxy.echo2(null, echoRequest2);
      assertTrue(Arrays.equals(echoResponse2.getMessageList().toArray(),
          new String[]{}));

      TestProtos.AddRequestProto addRequest =
          TestProtos.AddRequestProto.newBuilder().setParam1(1)
              .setParam2(2).build();
      TestProtos.AddResponseProto addResponse =
          proxy.add(null, addRequest);
      assertEquals(addResponse.getResult(), 3);

      Integer[] integers = new Integer[] {1, 2};
      TestProtos.AddRequestProto2 addRequest2 =
          TestProtos.AddRequestProto2.newBuilder().addAllParams(
              Arrays.asList(integers)).build();
      addResponse = proxy.add2(null, addRequest2);
      assertEquals(addResponse.getResult(), 3);

      boolean caught = false;
      try {
        proxy.error(null, newEmptyRequest());
      } catch (ServiceException e) {
        if(LOG.isDebugEnabled()) {
          LOG.debug("Caught " + e);
        }
        caught = true;
      }
      assertTrue(caught);
      rb = getMetrics(server.rpcDetailedMetrics.name());
      assertCounter("RpcServerExceptionNumOps", 1L, rb);

      //proxy.testServerGet();

      // create multiple threads and make them do large data transfers
      System.out.println("Starting multi-threaded RPC test...");
      server.setSocketSendBufSize(1024);
      Thread threadId[] = new Thread[numThreads];
      for (int i = 0; i < numThreads; i++) {
        Transactions trans = new Transactions(proxy, datasize);
        threadId[i] = new Thread(trans, "TransactionThread-" + i);
        threadId[i].start();
      }

      // wait for all transactions to get over
      System.out.println("Waiting for all threads to finish RPCs...");
      for (int i = 0; i < numThreads; i++) {
        try {
          threadId[i].join();
        } catch (InterruptedException e) {
          i--;      // retry
        }
      }
    } finally {
      stop(server, proxy);
    }
  }

  @Test
  public void testClientWithoutServer() throws Exception {
    TestRpcService proxy;

    short invalidPort = 20;
    InetSocketAddress invalidAddress = new InetSocketAddress(ADDRESS,
        invalidPort);
    long invalidClientVersion = 1L;
    try {
      proxy = RPC.getProxy(TestRpcService.class,
          invalidClientVersion, invalidAddress, conf);
      // Test echo method
      proxy.echo(null, newEchoRequest("hello"));
      fail("We should not have reached here");
    } catch (ServiceException ioe) {
      //this is what we expected
      if (!(ioe.getCause() instanceof ConnectException)) {
        fail("We should not have reached here");
      }
    }
  }

  private static final String ACL_CONFIG = "test.protocol.acl";

  private static class TestPolicyProvider extends PolicyProvider {

    @Override
    public Service[] getServices() {
      return new Service[] { new Service(ACL_CONFIG, TestRpcService.class) };
    }
  }

  private void doRPCs(Configuration myConf, boolean expectFailure) throws Exception {
    Server server;
    TestRpcService proxy = null;

    server = setupTestServer(myConf, 5);

    server.refreshServiceAcl(myConf, new TestPolicyProvider());

    TestProtos.EmptyRequestProto emptyRequestProto =
        TestProtos.EmptyRequestProto.newBuilder().build();

    try {
      proxy = getClient(addr, conf);
      proxy.ping(null, emptyRequestProto);
      if (expectFailure) {
        fail("Expect RPC.getProxy to fail with AuthorizationException!");
      }
    } catch (ServiceException e) {
      if (expectFailure) {
        RemoteException re = (RemoteException) e.getCause();
        assertTrue(re.unwrapRemoteException() instanceof AuthorizationException);
        assertEquals("RPC error code should be UNAUTHORIZED",
            RpcErrorCodeProto.FATAL_UNAUTHORIZED, re.getErrorCode());
      } else {
        throw e;
      }
    } finally {
      MetricsRecordBuilder rb = getMetrics(server.rpcMetrics.name());
      if (expectFailure) {
        assertCounter("RpcAuthorizationFailures", 1L, rb);
      } else {
        assertCounter("RpcAuthorizationSuccesses", 1L, rb);
      }
      //since we don't have authentication turned ON, we should see 
      // 0 for the authentication successes and 0 for failure
      assertCounter("RpcAuthenticationFailures", 0L, rb);
      assertCounter("RpcAuthenticationSuccesses", 0L, rb);

      stop(server, proxy);
    }
  }

  @Test
  public void testServerAddress() throws IOException {
    Server server;

    server = setupTestServer(conf, 5);
    try {
      InetSocketAddress bindAddr = NetUtils.getConnectAddress(server);
      assertEquals(InetAddress.getLocalHost(), bindAddr.getAddress());
    } finally {
      stop(server, null);
    }
  }

  @Test
  public void testAuthorization() throws Exception {
    Configuration myConf = new Configuration();
    myConf.setBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
        true);

    // Expect to succeed
    myConf.set(ACL_CONFIG, "*");
    doRPCs(myConf, false);

    // Reset authorization to expect failure
    myConf.set(ACL_CONFIG, "invalid invalid");
    doRPCs(myConf, true);

    myConf.setInt(CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY, 2);
    // Expect to succeed
    myConf.set(ACL_CONFIG, "*");
    doRPCs(myConf, false);

    // Reset authorization to expect failure
    myConf.set(ACL_CONFIG, "invalid invalid");
    doRPCs(myConf, true);
  }

  /**
   * Switch off setting socketTimeout values on RPC sockets.
   * Verify that RPC calls still work ok.
   */
  public void testNoPings() throws Exception {
    Configuration conf = new Configuration();

    conf.setBoolean("ipc.client.ping", false);
    new TestRPC().testCallsInternal(conf);

    conf.setInt(CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY, 2);
    new TestRPC().testCallsInternal(conf);
  }

  /**
   * Test stopping a non-registered proxy
   * @throws IOException
   */
  @Test(expected=HadoopIllegalArgumentException.class)
  public void testStopNonRegisteredProxy() throws IOException {
    RPC.stopProxy(null);
  }

  /**
   * Test that the mockProtocol helper returns mock proxies that can
   * be stopped without error.
   */
  @Test
  public void testStopMockObject() throws IOException {
    RPC.stopProxy(MockitoUtil.mockProtocol(TestProtocol.class));
  }

  @Test
  public void testStopProxy() throws IOException {
    RPC.setProtocolEngine(conf,
        StoppedProtocol.class, StoppedRpcEngine.class);

    StoppedProtocol proxy = RPC.getProxy(StoppedProtocol.class,
        StoppedProtocol.versionID, null, conf);
    StoppedInvocationHandler invocationHandler = (StoppedInvocationHandler)
        Proxy.getInvocationHandler(proxy);
    assertEquals(0, invocationHandler.getCloseCalled());
    RPC.stopProxy(proxy);
    assertEquals(1, invocationHandler.getCloseCalled());
  }

  @Test
  public void testWrappedStopProxy() throws IOException {
    StoppedProtocol wrappedProxy = RPC.getProxy(StoppedProtocol.class,
        StoppedProtocol.versionID, null, conf);
    StoppedInvocationHandler invocationHandler = (StoppedInvocationHandler)
        Proxy.getInvocationHandler(wrappedProxy);

    StoppedProtocol proxy = (StoppedProtocol) RetryProxy.create(
        StoppedProtocol.class, wrappedProxy, RetryPolicies.RETRY_FOREVER);

    assertEquals(0, invocationHandler.getCloseCalled());
    RPC.stopProxy(proxy);
    assertEquals(1, invocationHandler.getCloseCalled());
  }

  @Test
  public void testErrorMsgForInsecureClient() throws IOException {
    Server server;
    TestRpcService proxy = null;

    Configuration serverConf = new Configuration(conf);
    SecurityUtil.setAuthenticationMethod(AuthenticationMethod.KERBEROS,
        serverConf);
    UserGroupInformation.setConfiguration(serverConf);

    server = setupTestServer(serverConf, 5);

    boolean succeeded = false;

    try {
      UserGroupInformation.setConfiguration(conf);
      proxy = getClient(addr, conf);
      proxy.echo(null, newEchoRequest(""));
    } catch (ServiceException e) {
      assertTrue(e.getCause() instanceof RemoteException);
      RemoteException re = (RemoteException) e.getCause();
      LOG.info("LOGGING MESSAGE: " + re.getLocalizedMessage());
      assertEquals("RPC error code should be UNAUTHORIZED",
          RpcErrorCodeProto.FATAL_UNAUTHORIZED, re.getErrorCode());
      assertTrue(re.unwrapRemoteException() instanceof AccessControlException);
      succeeded = true;
    } finally {
      stop(server, proxy);
    }
    assertTrue(succeeded);

    conf.setInt(CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY, 2);

    UserGroupInformation.setConfiguration(serverConf);
    server = setupTestServer(serverConf, 5);
    succeeded = false;
    proxy = null;
    try {
      UserGroupInformation.setConfiguration(conf);
      proxy = getClient(addr, conf);
      proxy.echo(null, newEchoRequest(""));
    } catch (ServiceException e) {
      RemoteException re = (RemoteException) e.getCause();
      LOG.info("LOGGING MESSAGE: " + re.getLocalizedMessage());
      assertEquals("RPC error code should be UNAUTHORIZED",
          RpcErrorCodeProto.FATAL_UNAUTHORIZED, re.getErrorCode());
      assertTrue(re.unwrapRemoteException() instanceof AccessControlException);
      succeeded = true;
    } finally {
      stop(server, proxy);
    }
    assertTrue(succeeded);
  }

  /**
   * Test that server.stop() properly stops all threads
   */
  @Test
  public void testStopsAllThreads() throws IOException, InterruptedException {
    Server server;

    int threadsBefore = countThreads("Server$Listener$Reader");
    assertEquals("Expect no Reader threads running before test",
        0, threadsBefore);

    server = setupTestServer(conf, 5);

    try {
      // Wait for at least one reader thread to start
      int threadsRunning = 0;
      long totalSleepTime = 0;
      do {
        totalSleepTime += 10;
        Thread.sleep(10);
        threadsRunning = countThreads("Server$Listener$Reader");
      } while (threadsRunning == 0 && totalSleepTime < 5000);

      // Validate that at least one thread started (we didn't timeout)
      threadsRunning = countThreads("Server$Listener$Reader");
      assertTrue(threadsRunning > 0);
    } finally {
      server.stop();
    }

    int threadsAfter = countThreads("Server$Listener$Reader");
    assertEquals("Expect no Reader threads left running after test",
        0, threadsAfter);
  }

  @Test
  public void testRPCBuilder() throws IOException {
    // Test mandatory field conf
    try {
      new RPC.Builder(null).setProtocol(TestProtocol.class)
          .setInstance(new TestImpl()).setBindAddress(ADDRESS).setPort(0)
          .setNumHandlers(5).setVerbose(true).build();
      fail("Didn't throw HadoopIllegalArgumentException");
    } catch (Exception e) {
      if (!(e instanceof HadoopIllegalArgumentException)) {
        fail("Expecting HadoopIllegalArgumentException but caught " + e);
      }
    }
    // Test mandatory field protocol
    try {
      new RPC.Builder(conf).setInstance(new TestImpl()).setBindAddress(ADDRESS)
          .setPort(0).setNumHandlers(5).setVerbose(true).build();
      fail("Didn't throw HadoopIllegalArgumentException");
    } catch (Exception e) {
      if (!(e instanceof HadoopIllegalArgumentException)) {
        fail("Expecting HadoopIllegalArgumentException but caught " + e);
      }
    }
    // Test mandatory field instance
    try {
      new RPC.Builder(conf).setProtocol(TestProtocol.class)
          .setBindAddress(ADDRESS).setPort(0).setNumHandlers(5)
          .setVerbose(true).build();
      fail("Didn't throw HadoopIllegalArgumentException");
    } catch (Exception e) {
      if (!(e instanceof HadoopIllegalArgumentException)) {
        fail("Expecting HadoopIllegalArgumentException but caught " + e);
      }
    }
  }

  @Test(timeout=90000)
  public void testRPCInterruptedSimple() throws Exception {
    Server server;
    TestRpcService proxy = null;

    RPC.Builder builder = newServerBuilder(conf)
        .setNumHandlers(5).setVerbose(true)
        .setSecretManager(null);

    server = setupTestServer(builder);

    try {
      proxy = getClient(addr, conf);
      // Connect to the server

      proxy.ping(null, newEmptyRequest());
      // Interrupt self, try another call
      Thread.currentThread().interrupt();
      try {
        proxy.ping(null, newEmptyRequest());
        fail("Interruption did not cause IPC to fail");
      } catch (ServiceException se) {
        if (se.toString().contains("InterruptedException") ||
            se.getCause() instanceof InterruptedIOException) {
          // clear interrupt status for future tests
          Thread.interrupted();
          return;
        }
        throw se;
      }
    } finally {
      stop(server, proxy);
    }
  }

  @Test(timeout=30000)
  public void testRPCInterrupted() throws Exception {
    Server server;

    RPC.Builder builder = newServerBuilder(conf)
        .setNumHandlers(5).setVerbose(true)
        .setSecretManager(null);
    server = setupTestServer(builder);

    int numConcurrentRPC = 200;
    final CyclicBarrier barrier = new CyclicBarrier(numConcurrentRPC);
    final CountDownLatch latch = new CountDownLatch(numConcurrentRPC);
    final AtomicBoolean leaderRunning = new AtomicBoolean(true);
    final AtomicReference<Throwable> error = new AtomicReference<>();
    Thread leaderThread = null;

    try {
      for (int i = 0; i < numConcurrentRPC; i++) {
        final int num = i;
        final TestRpcService proxy = getClient(addr, conf);
        Thread rpcThread = new Thread(new Runnable() {
          @Override
          public void run() {
            try {
              barrier.await();
              while (num == 0 || leaderRunning.get()) {
                proxy.slowPing(null, newSlowPingRequest(false));
              }

              proxy.slowPing(null, newSlowPingRequest(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;
        }
      }
      // 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);
    } finally {
      server.stop();
    }
    // 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();
  }

  @Test
  public void testConnectionPing() throws Exception {
    Server server;
    TestRpcService proxy = null;

    int pingInterval = 50;
    conf.setBoolean(CommonConfigurationKeys.IPC_CLIENT_PING_KEY, true);
    conf.setInt(CommonConfigurationKeys.IPC_PING_INTERVAL_KEY, pingInterval);
    server = setupTestServer(conf, 5);

    try {
      proxy = getClient(addr, conf);

      proxy.sleep(null, newSleepRequest(pingInterval * 4));
    } finally {
      stop(server, proxy);
    }
  }

  @Test(timeout=30000)
  public void testExternalCall() throws Exception {
    final UserGroupInformation ugi = UserGroupInformation
        .createUserForTesting("user123", new String[0]);
    final IOException expectedIOE = new IOException("boom");

    // use 1 handler so the callq can be plugged
    final Server server = setupTestServer(conf, 1);
    try {
      final AtomicBoolean result = new AtomicBoolean();

      ExternalCall<String> remoteUserCall = newExtCall(ugi,
          new PrivilegedExceptionAction<String>() {
            @Override
            public String run() throws Exception {
              return UserGroupInformation.getCurrentUser().getUserName();
            }
          });

      ExternalCall<String> exceptionCall = newExtCall(ugi,
          new PrivilegedExceptionAction<String>() {
            @Override
            public String run() throws Exception {
              throw expectedIOE;
            }
          });

      final CountDownLatch latch = new CountDownLatch(1);
      final CyclicBarrier barrier = new CyclicBarrier(2);

      ExternalCall<Void> barrierCall = newExtCall(ugi,
          new PrivilegedExceptionAction<Void>() {
            @Override
            public Void run() throws Exception {
              // notify we are in a handler and then wait to keep the callq
              // plugged up
              latch.countDown();
              barrier.await();
              return null;
            }
          });

      server.queueCall(barrierCall);
      server.queueCall(exceptionCall);
      server.queueCall(remoteUserCall);

      // wait for barrier call to enter the handler, check that the other 2
      // calls are actually queued
      latch.await();
      assertEquals(2, server.getCallQueueLen());

      // unplug the callq
      barrier.await();
      barrierCall.get();

      // verify correct ugi is used
      String answer = remoteUserCall.get();
      assertEquals(ugi.getUserName(), answer);

      try {
        exceptionCall.get();
        fail("didn't throw");
      } catch (ExecutionException ee) {
        assertTrue((ee.getCause()) instanceof IOException);
        assertEquals(expectedIOE.getMessage(), ee.getCause().getMessage());
      }
    } finally {
      server.stop();
    }
  }

  private <T> ExternalCall<T> newExtCall(final UserGroupInformation ugi,
      PrivilegedExceptionAction<T> callable) {
    return new ExternalCall<T>(callable) {
      @Override
      public String getProtocol() {
        return "test";
      }
      @Override
      public UserGroupInformation getRemoteUser() {
        return ugi;
      }
    };
  }

  @Test
  public void testRpcMetrics() throws Exception {
    final Server server;
    TestRpcService proxy = null;

    final int interval = 1;
    conf.setBoolean(CommonConfigurationKeys.
        RPC_METRICS_QUANTILE_ENABLE, true);
    conf.set(CommonConfigurationKeys.
        RPC_METRICS_PERCENTILES_INTERVALS_KEY, "" + interval);

    server = setupTestServer(conf, 5);
    String testUser = "testUser";
    UserGroupInformation anotherUser =
        UserGroupInformation.createRemoteUser(testUser);
    TestRpcService proxy2 =
        anotherUser.doAs(new PrivilegedAction<TestRpcService>() {
          public TestRpcService run() {
            try {
              return RPC.getProxy(TestRpcService.class, 0,
                  server.getListenerAddress(), conf);
            } catch (IOException e) {
              e.printStackTrace();
            }
            return null;
          }
        });
    try {
      proxy = getClient(addr, conf);

      for (int i = 0; i < 1000; i++) {
        proxy.ping(null, newEmptyRequest());

        proxy.echo(null, newEchoRequest("" + i));
        proxy2.echo(null, newEchoRequest("" + i));
      }
      MetricsRecordBuilder rpcMetrics =
          getMetrics(server.getRpcMetrics().name());
      assertEquals("Expected correct rpc queue count",
          3000, getLongCounter("RpcQueueTimeNumOps", rpcMetrics));
      assertEquals("Expected correct rpc processing count",
          3000, getLongCounter("RpcProcessingTimeNumOps", rpcMetrics));
      assertEquals("Expected correct rpc lock wait count",
          3000, getLongCounter("RpcLockWaitTimeNumOps", rpcMetrics));
      assertEquals("Expected zero rpc lock wait time",
          0, getDoubleGauge("RpcLockWaitTimeAvgTime", rpcMetrics), 0.001);
      MetricsAsserts.assertQuantileGauges("RpcQueueTime" + interval + "s",
          rpcMetrics);
      MetricsAsserts.assertQuantileGauges("RpcProcessingTime" + interval + "s",
          rpcMetrics);
      String actualUserVsCon = MetricsAsserts
          .getStringMetric("NumOpenConnectionsPerUser", rpcMetrics);
      String proxyUser =
          UserGroupInformation.getCurrentUser().getShortUserName();
      assertTrue(actualUserVsCon.contains("\"" + proxyUser + "\":1"));
      assertTrue(actualUserVsCon.contains("\"" + testUser + "\":1"));

      proxy.lockAndSleep(null, newSleepRequest(5));
      rpcMetrics = getMetrics(server.getRpcMetrics().name());
      assertGauge("RpcLockWaitTimeAvgTime", 10000.0, rpcMetrics);
    } finally {
      if (proxy2 != null) {
        RPC.stopProxy(proxy2);
      }
      stop(server, proxy);
    }
  }

  /**
   *  Test RPC backoff by queue full.
   */
  @Test (timeout=30000)
  public void testClientBackOff() throws Exception {
    Server server;
    final TestRpcService proxy;

    boolean succeeded = false;
    final int numClients = 2;
    final List<Future<Void>> res = new ArrayList<Future<Void>>();
    final ExecutorService executorService =
        Executors.newFixedThreadPool(numClients);
    conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 0);
    conf.setBoolean(CommonConfigurationKeys.IPC_NAMESPACE +
        ".0." + CommonConfigurationKeys.IPC_BACKOFF_ENABLE, true);
    RPC.Builder builder = newServerBuilder(conf)
        .setQueueSizePerHandler(1).setNumHandlers(1).setVerbose(true);
    server = setupTestServer(builder);

    @SuppressWarnings("unchecked")
    CallQueueManager<Call> spy = spy((CallQueueManager<Call>) Whitebox
        .getInternalState(server, "callQueue"));
    Whitebox.setInternalState(server, "callQueue", spy);

    Exception lastException = null;
    proxy = getClient(addr, conf);
    try {
      // start a sleep RPC call to consume the only handler thread.
      // Start another sleep RPC call to make callQueue full.
      // Start another sleep RPC call to make reader thread block on CallQueue.
      for (int i = 0; i < numClients; i++) {
        res.add(executorService.submit(
            new Callable<Void>() {
              @Override
              public Void call() throws ServiceException, InterruptedException {
                proxy.sleep(null, newSleepRequest(100000));
                return null;
              }
            }));
        verify(spy, timeout(500).times(i + 1)).addInternal(
            Mockito.<Call>anyObject(), Mockito.eq(false));
      }
      try {
        proxy.sleep(null, newSleepRequest(100));
      } catch (ServiceException e) {
        RemoteException re = (RemoteException) e.getCause();
        IOException unwrapExeption = re.unwrapRemoteException();
        if (unwrapExeption instanceof RetriableException) {
          succeeded = true;
        } else {
          lastException = unwrapExeption;
        }
      }
    } finally {
      executorService.shutdown();
      stop(server, proxy);
    }
    if (lastException != null) {
      LOG.error("Last received non-RetriableException:", lastException);
    }
    assertTrue("RetriableException not received", succeeded);
  }

  /**
   *  Test RPC backoff by response time of each priority level.
   */
  @Test (timeout=30000)
  public void testClientBackOffByResponseTime() throws Exception {
    final TestRpcService proxy;
    boolean succeeded = false;
    final int numClients = 1;

    GenericTestUtils.setLogLevel(DecayRpcScheduler.LOG, Level.DEBUG);
    GenericTestUtils.setLogLevel(RPC.LOG, Level.DEBUG);

    final List<Future<Void>> res = new ArrayList<Future<Void>>();
    final ExecutorService executorService =
        Executors.newFixedThreadPool(numClients);
    conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 0);

    final String ns = CommonConfigurationKeys.IPC_NAMESPACE + ".0";
    Server server = setupDecayRpcSchedulerandTestServer(ns + ".");

    @SuppressWarnings("unchecked")
    CallQueueManager<Call> spy = spy((CallQueueManager<Call>) Whitebox
        .getInternalState(server, "callQueue"));
    Whitebox.setInternalState(server, "callQueue", spy);

    Exception lastException = null;
    proxy = getClient(addr, conf);

    try {
      // start a sleep RPC call that sleeps 3s.
      for (int i = 0; i < numClients; i++) {
        res.add(executorService.submit(
            new Callable<Void>() {
              @Override
              public Void call() throws ServiceException, InterruptedException {
                proxy.sleep(null, newSleepRequest(3000));
                return null;
              }
            }));
        verify(spy, timeout(500).times(i + 1)).addInternal(
            Mockito.<Call>anyObject(), Mockito.eq(false));
      }
      // Start another sleep RPC call and verify the call is backed off due to
      // avg response time(3s) exceeds threshold (2s).
      try {
        // wait for the 1st response time update
        Thread.sleep(5500);
        proxy.sleep(null, newSleepRequest(100));
      } catch (ServiceException e) {
        RemoteException re = (RemoteException) e.getCause();
        IOException unwrapExeption = re.unwrapRemoteException();
        if (unwrapExeption instanceof RetriableException) {
          succeeded = true;
        } else {
          lastException = unwrapExeption;
        }
      }
    } finally {
      executorService.shutdown();
      stop(server, proxy);
    }
    if (lastException != null) {
      LOG.error("Last received non-RetriableException:", lastException);
    }
    assertTrue("RetriableException not received", succeeded);
  }

  /** Test that the metrics for DecayRpcScheduler are updated. */
  @Test (timeout=30000)
  public void testDecayRpcSchedulerMetrics() throws Exception {
    final String ns = CommonConfigurationKeys.IPC_NAMESPACE + ".0";
    Server server = setupDecayRpcSchedulerandTestServer(ns + ".");

    MetricsRecordBuilder rb1 =
        getMetrics("DecayRpcSchedulerMetrics2." + ns);
    final long beginDecayedCallVolume = MetricsAsserts.getLongCounter(
        "DecayedCallVolume", rb1);
    final long beginRawCallVolume = MetricsAsserts.getLongCounter(
        "CallVolume", rb1);
    final int beginUniqueCaller = MetricsAsserts.getIntCounter("UniqueCallers",
        rb1);

    TestRpcService proxy = getClient(addr, conf);
    try {
      for (int i = 0; i < 2; i++) {
        proxy.sleep(null, newSleepRequest(100));
      }

      // Lets Metric system update latest metrics
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          MetricsRecordBuilder rb2 =
              getMetrics("DecayRpcSchedulerMetrics2." + ns);
          long decayedCallVolume1 = MetricsAsserts.getLongCounter(
              "DecayedCallVolume", rb2);
          long rawCallVolume1 = MetricsAsserts.getLongCounter(
              "CallVolume", rb2);
          int uniqueCaller1 = MetricsAsserts.getIntCounter(
              "UniqueCallers", rb2);
          long callVolumePriority0 = MetricsAsserts.getLongGauge(
              "Priority.0.CompletedCallVolume", rb2);
          long callVolumePriority1 = MetricsAsserts.getLongGauge(
              "Priority.1.CompletedCallVolume", rb2);
          double avgRespTimePriority0 = MetricsAsserts.getDoubleGauge(
              "Priority.0.AvgResponseTime", rb2);
          double avgRespTimePriority1 = MetricsAsserts.getDoubleGauge(
              "Priority.1.AvgResponseTime", rb2);

          LOG.info("DecayedCallVolume: {}", decayedCallVolume1);
          LOG.info("CallVolume: {}", rawCallVolume1);
          LOG.info("UniqueCaller: {}", uniqueCaller1);
          LOG.info("Priority.0.CompletedCallVolume: {}", callVolumePriority0);
          LOG.info("Priority.1.CompletedCallVolume: {}", callVolumePriority1);
          LOG.info("Priority.0.AvgResponseTime: {}", avgRespTimePriority0);
          LOG.info("Priority.1.AvgResponseTime: {}", avgRespTimePriority1);

          return decayedCallVolume1 > beginDecayedCallVolume &&
              rawCallVolume1 > beginRawCallVolume &&
              uniqueCaller1 > beginUniqueCaller;
        }
      }, 30, 60000);
    } finally {
      stop(server, proxy);
    }
  }

  private Server setupDecayRpcSchedulerandTestServer(String ns)
      throws Exception {
    final int queueSizePerHandler = 3;

    conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 0);
    conf.setBoolean(ns + CommonConfigurationKeys.IPC_BACKOFF_ENABLE, true);
    conf.setStrings(ns + CommonConfigurationKeys.IPC_CALLQUEUE_IMPL_KEY,
        "org.apache.hadoop.ipc.FairCallQueue");
    conf.setStrings(ns + CommonConfigurationKeys.IPC_SCHEDULER_IMPL_KEY,
        "org.apache.hadoop.ipc.DecayRpcScheduler");
    conf.setInt(ns + CommonConfigurationKeys.IPC_SCHEDULER_PRIORITY_LEVELS_KEY,
        2);
    conf.setBoolean(ns +
            DecayRpcScheduler.IPC_DECAYSCHEDULER_BACKOFF_RESPONSETIME_ENABLE_KEY,
        true);
    // set a small thresholds 2s and 4s for level 0 and level 1 for testing
    conf.set(ns +
            DecayRpcScheduler.IPC_DECAYSCHEDULER_BACKOFF_RESPONSETIME_THRESHOLDS_KEY
        , "2s, 4s");

    // Set max queue size to 3 so that 2 calls from the test won't trigger
    // back off because the queue is full.
    RPC.Builder builder = newServerBuilder(conf)
        .setQueueSizePerHandler(queueSizePerHandler).setNumHandlers(1)
        .setVerbose(true);
    return setupTestServer(builder);
  }

  /**
   *  Test RPC timeout.
   */
  @Test(timeout=30000)
  public void testClientRpcTimeout() throws Exception {
    Server server;
    TestRpcService proxy = null;

    RPC.Builder builder = newServerBuilder(conf)
        .setQueueSizePerHandler(1).setNumHandlers(1).setVerbose(true);
    server = setupTestServer(builder);

    try {
      // Test RPC timeout with default ipc.client.ping.
      try {
        Configuration c = new Configuration(conf);
        c.setInt(CommonConfigurationKeys.IPC_CLIENT_RPC_TIMEOUT_KEY, 1000);
        proxy = getClient(addr, c);
        proxy.sleep(null, newSleepRequest(3000));
        fail("RPC should time out.");
      } catch (ServiceException e) {
        assertTrue(e.getCause() instanceof SocketTimeoutException);
        LOG.info("got expected timeout.", e);
      }

      // Test RPC timeout when ipc.client.ping is false.
      try {
        Configuration c = new Configuration(conf);
        c.setBoolean(CommonConfigurationKeys.IPC_CLIENT_PING_KEY, false);
        c.setInt(CommonConfigurationKeys.IPC_CLIENT_RPC_TIMEOUT_KEY, 1000);
        proxy = getClient(addr, c);
        proxy.sleep(null, newSleepRequest(3000));
        fail("RPC should time out.");
      } catch (ServiceException e) {
        assertTrue(e.getCause() instanceof SocketTimeoutException);
        LOG.info("got expected timeout.", e);
      }

      // Test negative timeout value.
      try {
        Configuration c = new Configuration(conf);
        c.setInt(CommonConfigurationKeys.IPC_CLIENT_RPC_TIMEOUT_KEY, -1);
        proxy = getClient(addr, c);
        proxy.sleep(null, newSleepRequest(2000));
      } catch (ServiceException e) {
        LOG.info("got unexpected exception.", e);
        fail("RPC should not time out.");
      }

      // Test RPC timeout greater than ipc.ping.interval.
      try {
        Configuration c = new Configuration(conf);
        c.setBoolean(CommonConfigurationKeys.IPC_CLIENT_PING_KEY, true);
        c.setInt(CommonConfigurationKeys.IPC_PING_INTERVAL_KEY, 800);
        c.setInt(CommonConfigurationKeys.IPC_CLIENT_RPC_TIMEOUT_KEY, 1000);
        proxy = getClient(addr, c);

        try {
          // should not time out because effective rpc-timeout is
          // multiple of ping interval: 1600 (= 800 * (1000 / 800 + 1))
          proxy.sleep(null, newSleepRequest(1300));
        } catch (ServiceException e) {
          LOG.info("got unexpected exception.", e);
          fail("RPC should not time out.");
        }

        proxy.sleep(null, newSleepRequest(2000));
        fail("RPC should time out.");
      } catch (ServiceException e) {
        assertTrue(e.getCause() instanceof SocketTimeoutException);
        LOG.info("got expected timeout.", e);
      }

    } finally {
      stop(server, proxy);
    }
  }

  public static class FakeRequestClass extends RpcWritable {
    static volatile IOException exception;
    @Override
    void writeTo(ResponseBuffer out) throws IOException {
      throw new UnsupportedOperationException();
    }
    @Override
    <T> T readFrom(ByteBuffer bb) throws IOException {
      throw exception;
    }
  }

  @SuppressWarnings("serial")
  public static class TestReaderException extends IOException {
    public TestReaderException(String msg) {
      super(msg);
    }
    @Override
    public boolean equals(Object t) {
      return (t.getClass() == TestReaderException.class) &&
             getMessage().equals(((TestReaderException)t).getMessage());
    }
  }

  @Test (timeout=30000)
  public void testReaderExceptions() throws Exception {
    Server server = null;
    TestRpcService proxy = null;

    // will attempt to return this exception from a reader with and w/o
    // the connection closing.
    IOException expectedIOE = new TestReaderException("testing123");

    @SuppressWarnings("serial")
    IOException rseError = new RpcServerException("keepalive", expectedIOE){
      @Override
      public RpcStatusProto getRpcStatusProto() {
        return RpcStatusProto.ERROR;
      }
    };
    @SuppressWarnings("serial")
    IOException rseFatal = new RpcServerException("disconnect", expectedIOE) {
      @Override
      public RpcStatusProto getRpcStatusProto() {
        return RpcStatusProto.FATAL;
      }
    };

    try {
      RPC.Builder builder = newServerBuilder(conf)
          .setQueueSizePerHandler(1).setNumHandlers(1).setVerbose(true);
      server = setupTestServer(builder);
      Whitebox.setInternalState(
          server, "rpcRequestClass", FakeRequestClass.class);
      MutableCounterLong authMetric =
          (MutableCounterLong)Whitebox.getInternalState(
              server.getRpcMetrics(), "rpcAuthorizationSuccesses");

      proxy = getClient(addr, conf);
      boolean isDisconnected = true;
      Connection lastConn = null;
      long expectedAuths = 0;

      // fuzz the client.
      for (int i=0; i < 128; i++) {
        String reqName = "request[" + i + "]";
        int r = ThreadLocalRandom.current().nextInt();
        final boolean doDisconnect = r % 4 == 0;
        LOG.info("TestDisconnect request[" + i + "] " +
                 " shouldConnect=" + isDisconnected +
                 " willDisconnect=" + doDisconnect);
        if (isDisconnected) {
          expectedAuths++;
        }
        try {
          FakeRequestClass.exception = doDisconnect ? rseFatal : rseError;
          proxy.ping(null, newEmptyRequest());
          fail(reqName + " didn't fail");
        } catch (ServiceException e) {
          RemoteException re = (RemoteException)e.getCause();
          assertEquals(reqName, expectedIOE, re.unwrapRemoteException());
        }
        // check authorizations to ensure new connection when expected,
        // then conclusively determine if connections are disconnected
        // correctly.
        assertEquals(reqName, expectedAuths, authMetric.value());
        if (!doDisconnect) {
          // if it wasn't fatal, verify there's only one open connection.
          Connection[] conns = server.getConnections();
          assertEquals(reqName, 1, conns.length);
          // verify whether the connection should have been reused.
          if (isDisconnected) {
            assertNotSame(reqName, lastConn, conns[0]);
          } else {
            assertSame(reqName, lastConn, conns[0]);
          }
          lastConn = conns[0];
        } else if (lastConn != null) {
          // avoid race condition in server where connection may not be
          // fully removed yet.  just make sure it's marked for being closed.
          // the open connection checks above ensure correct behavior.
          assertTrue(reqName, lastConn.shouldClose());
        }
        isDisconnected = doDisconnect;
      }
    } finally {
      stop(server, proxy);
    }
  }

  public static void main(String[] args) throws Exception {
    new TestRPC().testCallsInternal(conf);
  }
}