diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java index a120cd5..b22c0a2 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java @@ -24,7 +24,10 @@ import static org.mockito.Matchers.anyInt; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.spy; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; import java.io.IOException; +import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.net.Socket; @@ -36,9 +39,11 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.IpcProtocol; +import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.SmallTests; import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RpcRequestBody; +import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RpcResponseBody; import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.net.NetUtils; import org.apache.hadoop.util.StringUtils; @@ -48,6 +53,7 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import com.google.protobuf.ByteString; import com.google.protobuf.Message; @Category(SmallTests.class) @@ -63,7 +69,7 @@ public class TestIPC { public Message call(Class protocol, RpcRequestBody param, long receiveTime, MonitoredRPCHandler status) throws IOException { - return param; + return RpcResponseBody.newBuilder().setResponse(param.getRequest()).build(); } } @@ -95,4 +101,71 @@ public class TestIPC { assertTrue(StringUtils.stringifyException(e).contains("Injected fault")); } } + + /** + * A nothing protocol used in test below. + */ + interface NothingProtocol extends IpcProtocol { + RpcResponseBody doNothing(); + } + + public static class DoNothing implements NothingProtocol { + public RpcResponseBody doNothing() {return RpcResponseBody.newBuilder().build();} + } + + public static void main(String[] args) + throws IOException, SecurityException, NoSuchMethodException, InterruptedException { + if (args.length != 2) { + System.out.println("Usage: TestIPC "); + return; + } + // ((Log4JLogger)HBaseServer.LOG).getLogger().setLevel(Level.INFO); + // ((Log4JLogger)HBaseClient.LOG).getLogger().setLevel(Level.INFO); + int cycles = Integer.parseInt(args[0]); + int cellcount = Integer.parseInt(args[1]); + Configuration conf = HBaseConfiguration.create(); + TestRpcServer rpcServer = new TestRpcServer(); + HBaseClient client = new HBaseClient(conf, HConstants.CLUSTER_ID_DEFAULT); + RpcRequestBody.Builder builder = RpcRequestBody.newBuilder(); + byte [] xyz = new byte [] {'x', 'y', 'z'}; + KeyValue kv = new KeyValue(xyz, xyz, xyz, xyz); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DataOutputStream dos = new DataOutputStream(baos); + for (int i = 0; i < cellcount; i++) { + KeyValue.write(kv, dos); + } + dos.close(); + byte [] bytes = baos.toByteArray(); + Method m = NothingProtocol.class.getMethod("doNothing", null); + builder.setMethodName(m.getName()); + builder.setRequest(ByteString.copyFrom(bytes)); + RpcRequestBody body = builder.build(); + try { + rpcServer.start(); + InetSocketAddress address = rpcServer.getListenerAddress(); + // Get any method name... just so it is not null + long startTime = System.currentTimeMillis(); + User user = User.getCurrent(); + for (int i = 0; i < cycles; i++) { + if (i % 1000 == 0) { + LOG.info("" + i); + // Uncomment this for a thread dump every so often. + // ReflectionUtils.printThreadInfo(new PrintWriter(System.out), + // "Thread dump " + Thread.currentThread().getName()); + } + Message response = client.call(body, address, NothingProtocol.class, user, 0); + /* + int count = 0; + while (p.getSecond().advance()) { + count++; + } + assertEquals(cells.size(), count);*/ + } + LOG.info("Cycled " + cycles + " time(s) with " + cellcount + " cell(s) in " + + (System.currentTimeMillis() - startTime) + "ms"); + } finally { + client.stop(); + rpcServer.stop(); + } + } } \ No newline at end of file