@Test public void testSimpleCall() { RpcServerInterface mockRpcServer = Mockito.mock(RpcServerInterface.class); Mockito.when(mockRpcServer.isStarted()).thenReturn(true); RpcServer.Call mockCall = Mockito.mock(RpcServer.Call.class); mockCall.connection = Mockito.mock(RpcServer.Connection.class); CallRunner cr = new CallRunner(mockRpcServer, mockCall, new UserProvider()); cr.run(); }
public void run() { try { if (!call.connection.channel.isOpen()) { if (RpcServer.LOG.isDebugEnabled()) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call); } return; } this.status.setStatus("Setting up call"); this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort()); if (RpcServer.LOG.isDebugEnabled()) { UserGroupInformation remoteUser = call.connection.user; RpcServer.LOG.debug(call.toShortString() + " executing as " + ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName())); } Throwable errorThrowable = null; String error = null; Pair<Message, CellScanner> resultPair = null; RpcServer.CurCall.set(call); TraceScope traceScope = null; try { if (!this.rpcServer.isStarted()) { throw new ServerNotRunningYetException("Server is not running yet"); } if (call.tinfo != null) { traceScope = Trace.startSpan(call.toTraceString(), call.tinfo); } RequestContext.set(userProvider.create(call.connection.user), RpcServer.getRemoteIp(), call.connection.service); resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner, call.timestamp, this.status); } catch (Throwable e) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e); errorThrowable = e; error = StringUtils.stringifyException(e); } finally { if (traceScope != null) { traceScope.close(); } RequestContext.clear(); } RpcServer.CurCall.set(null); this.rpcServer.addCallSize(call.getSize() * -1); if (!call.isDelayed() || !call.isReturnValueDelayed()) { Message param = resultPair != null ? resultPair.getFirst() : null; CellScanner cells = resultPair != null ? resultPair.getSecond() : null; call.setResponse(param, cells, errorThrowable, error); } call.sendResponseIfReady(); this.status.markComplete("Sent response"); this.status.pause("Waiting for a call"); } catch (OutOfMemoryError e) { if (this.rpcServer.getErrorHandler() != null) { if (this.rpcServer.getErrorHandler().checkOOME(e)) { RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError"); return; } } else { throw e; } } catch (ClosedChannelException cce) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " + "this means that the server was processing a " + "request but the client went away. The error message was: " + cce.getMessage()); } catch (Exception e) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught: " + StringUtils.stringifyException(e)); } }
CallRunner { public void run() { try { if (!call.connection.channel.isOpen()) { if (RpcServer.LOG.isDebugEnabled()) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call); } return; } this.status.setStatus("Setting up call"); this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort()); if (RpcServer.LOG.isDebugEnabled()) { UserGroupInformation remoteUser = call.connection.user; RpcServer.LOG.debug(call.toShortString() + " executing as " + ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName())); } Throwable errorThrowable = null; String error = null; Pair<Message, CellScanner> resultPair = null; RpcServer.CurCall.set(call); TraceScope traceScope = null; try { if (!this.rpcServer.isStarted()) { throw new ServerNotRunningYetException("Server is not running yet"); } if (call.tinfo != null) { traceScope = Trace.startSpan(call.toTraceString(), call.tinfo); } RequestContext.set(userProvider.create(call.connection.user), RpcServer.getRemoteIp(), call.connection.service); resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner, call.timestamp, this.status); } catch (Throwable e) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e); errorThrowable = e; error = StringUtils.stringifyException(e); } finally { if (traceScope != null) { traceScope.close(); } RequestContext.clear(); } RpcServer.CurCall.set(null); this.rpcServer.addCallSize(call.getSize() * -1); if (!call.isDelayed() || !call.isReturnValueDelayed()) { Message param = resultPair != null ? resultPair.getFirst() : null; CellScanner cells = resultPair != null ? resultPair.getSecond() : null; call.setResponse(param, cells, errorThrowable, error); } call.sendResponseIfReady(); this.status.markComplete("Sent response"); this.status.pause("Waiting for a call"); } catch (OutOfMemoryError e) { if (this.rpcServer.getErrorHandler() != null) { if (this.rpcServer.getErrorHandler().checkOOME(e)) { RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError"); return; } } else { throw e; } } catch (ClosedChannelException cce) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " + "this means that the server was processing a " + "request but the client went away. The error message was: " + cce.getMessage()); } catch (Exception e) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught: " + StringUtils.stringifyException(e)); } } }
CallRunner { public void run() { try { if (!call.connection.channel.isOpen()) { if (RpcServer.LOG.isDebugEnabled()) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call); } return; } this.status.setStatus("Setting up call"); this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort()); if (RpcServer.LOG.isDebugEnabled()) { UserGroupInformation remoteUser = call.connection.user; RpcServer.LOG.debug(call.toShortString() + " executing as " + ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName())); } Throwable errorThrowable = null; String error = null; Pair<Message, CellScanner> resultPair = null; RpcServer.CurCall.set(call); TraceScope traceScope = null; try { if (!this.rpcServer.isStarted()) { throw new ServerNotRunningYetException("Server is not running yet"); } if (call.tinfo != null) { traceScope = Trace.startSpan(call.toTraceString(), call.tinfo); } RequestContext.set(userProvider.create(call.connection.user), RpcServer.getRemoteIp(), call.connection.service); resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner, call.timestamp, this.status); } catch (Throwable e) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e); errorThrowable = e; error = StringUtils.stringifyException(e); } finally { if (traceScope != null) { traceScope.close(); } RequestContext.clear(); } RpcServer.CurCall.set(null); this.rpcServer.addCallSize(call.getSize() * -1); if (!call.isDelayed() || !call.isReturnValueDelayed()) { Message param = resultPair != null ? resultPair.getFirst() : null; CellScanner cells = resultPair != null ? resultPair.getSecond() : null; call.setResponse(param, cells, errorThrowable, error); } call.sendResponseIfReady(); this.status.markComplete("Sent response"); this.status.pause("Waiting for a call"); } catch (OutOfMemoryError e) { if (this.rpcServer.getErrorHandler() != null) { if (this.rpcServer.getErrorHandler().checkOOME(e)) { RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError"); return; } } else { throw e; } } catch (ClosedChannelException cce) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " + "this means that the server was processing a " + "request but the client went away. The error message was: " + cce.getMessage()); } catch (Exception e) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught: " + StringUtils.stringifyException(e)); } } CallRunner(final RpcServerInterface rpcServer, final Call call, UserProvider userProvider); }
CallRunner { public void run() { try { if (!call.connection.channel.isOpen()) { if (RpcServer.LOG.isDebugEnabled()) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call); } return; } this.status.setStatus("Setting up call"); this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort()); if (RpcServer.LOG.isDebugEnabled()) { UserGroupInformation remoteUser = call.connection.user; RpcServer.LOG.debug(call.toShortString() + " executing as " + ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName())); } Throwable errorThrowable = null; String error = null; Pair<Message, CellScanner> resultPair = null; RpcServer.CurCall.set(call); TraceScope traceScope = null; try { if (!this.rpcServer.isStarted()) { throw new ServerNotRunningYetException("Server is not running yet"); } if (call.tinfo != null) { traceScope = Trace.startSpan(call.toTraceString(), call.tinfo); } RequestContext.set(userProvider.create(call.connection.user), RpcServer.getRemoteIp(), call.connection.service); resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner, call.timestamp, this.status); } catch (Throwable e) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e); errorThrowable = e; error = StringUtils.stringifyException(e); } finally { if (traceScope != null) { traceScope.close(); } RequestContext.clear(); } RpcServer.CurCall.set(null); this.rpcServer.addCallSize(call.getSize() * -1); if (!call.isDelayed() || !call.isReturnValueDelayed()) { Message param = resultPair != null ? resultPair.getFirst() : null; CellScanner cells = resultPair != null ? resultPair.getSecond() : null; call.setResponse(param, cells, errorThrowable, error); } call.sendResponseIfReady(); this.status.markComplete("Sent response"); this.status.pause("Waiting for a call"); } catch (OutOfMemoryError e) { if (this.rpcServer.getErrorHandler() != null) { if (this.rpcServer.getErrorHandler().checkOOME(e)) { RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError"); return; } } else { throw e; } } catch (ClosedChannelException cce) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " + "this means that the server was processing a " + "request but the client went away. The error message was: " + cce.getMessage()); } catch (Exception e) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught: " + StringUtils.stringifyException(e)); } } CallRunner(final RpcServerInterface rpcServer, final Call call, UserProvider userProvider); Call getCall(); void run(); }
CallRunner { public void run() { try { if (!call.connection.channel.isOpen()) { if (RpcServer.LOG.isDebugEnabled()) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": skipped " + call); } return; } this.status.setStatus("Setting up call"); this.status.setConnection(call.connection.getHostAddress(), call.connection.getRemotePort()); if (RpcServer.LOG.isDebugEnabled()) { UserGroupInformation remoteUser = call.connection.user; RpcServer.LOG.debug(call.toShortString() + " executing as " + ((remoteUser == null) ? "NULL principal" : remoteUser.getUserName())); } Throwable errorThrowable = null; String error = null; Pair<Message, CellScanner> resultPair = null; RpcServer.CurCall.set(call); TraceScope traceScope = null; try { if (!this.rpcServer.isStarted()) { throw new ServerNotRunningYetException("Server is not running yet"); } if (call.tinfo != null) { traceScope = Trace.startSpan(call.toTraceString(), call.tinfo); } RequestContext.set(userProvider.create(call.connection.user), RpcServer.getRemoteIp(), call.connection.service); resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner, call.timestamp, this.status); } catch (Throwable e) { RpcServer.LOG.debug(Thread.currentThread().getName() + ": " + call.toShortString(), e); errorThrowable = e; error = StringUtils.stringifyException(e); } finally { if (traceScope != null) { traceScope.close(); } RequestContext.clear(); } RpcServer.CurCall.set(null); this.rpcServer.addCallSize(call.getSize() * -1); if (!call.isDelayed() || !call.isReturnValueDelayed()) { Message param = resultPair != null ? resultPair.getFirst() : null; CellScanner cells = resultPair != null ? resultPair.getSecond() : null; call.setResponse(param, cells, errorThrowable, error); } call.sendResponseIfReady(); this.status.markComplete("Sent response"); this.status.pause("Waiting for a call"); } catch (OutOfMemoryError e) { if (this.rpcServer.getErrorHandler() != null) { if (this.rpcServer.getErrorHandler().checkOOME(e)) { RpcServer.LOG.info(Thread.currentThread().getName() + ": exiting on OutOfMemoryError"); return; } } else { throw e; } } catch (ClosedChannelException cce) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " + "this means that the server was processing a " + "request but the client went away. The error message was: " + cce.getMessage()); } catch (Exception e) { RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught: " + StringUtils.stringifyException(e)); } } CallRunner(final RpcServerInterface rpcServer, final Call call, UserProvider userProvider); Call getCall(); void run();  }