| /** |
| * 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 static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| import static org.mockito.Matchers.anyInt; |
| import static org.mockito.Mockito.doAnswer; |
| import static org.mockito.Mockito.doThrow; |
| import static org.mockito.Mockito.mock; |
| import static org.mockito.Mockito.spy; |
| import static org.mockito.Mockito.timeout; |
| import static org.mockito.Mockito.verify; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.OutputStream; |
| import java.lang.reflect.Method; |
| import java.lang.reflect.Proxy; |
| import java.net.InetAddress; |
| import java.net.InetSocketAddress; |
| import java.net.ServerSocket; |
| import java.net.Socket; |
| import java.net.SocketAddress; |
| import java.net.SocketException; |
| import java.net.SocketTimeoutException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.concurrent.BrokenBarrierException; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.CyclicBarrier; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.TimeoutException; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import javax.net.SocketFactory; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.CommonConfigurationKeys; |
| import org.apache.hadoop.fs.CommonConfigurationKeysPublic; |
| import org.apache.hadoop.io.IOUtils; |
| import org.apache.hadoop.io.LongWritable; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.io.retry.DefaultFailoverProxyProvider; |
| import org.apache.hadoop.io.retry.FailoverProxyProvider; |
| import org.apache.hadoop.io.retry.Idempotent; |
| import org.apache.hadoop.io.retry.RetryPolicies; |
| import org.apache.hadoop.io.retry.RetryProxy; |
| import org.apache.hadoop.ipc.Client.ConnectionId; |
| import org.apache.hadoop.ipc.RPC.RpcKind; |
| import org.apache.hadoop.ipc.Server.Call; |
| import org.apache.hadoop.ipc.Server.Connection; |
| import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto; |
| import org.apache.hadoop.net.ConnectTimeoutException; |
| import org.apache.hadoop.net.NetUtils; |
| import org.apache.hadoop.security.KerberosInfo; |
| import org.apache.hadoop.security.SecurityUtil; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod; |
| import org.apache.hadoop.security.token.SecretManager.InvalidToken; |
| import org.apache.hadoop.test.GenericTestUtils; |
| import org.apache.hadoop.util.StringUtils; |
| import org.junit.Assert; |
| import org.junit.Assume; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.mockito.Mockito; |
| import org.mockito.internal.util.reflection.Whitebox; |
| import org.mockito.invocation.InvocationOnMock; |
| import org.mockito.stubbing.Answer; |
| |
| import com.google.common.base.Supplier; |
| import com.google.common.primitives.Bytes; |
| import com.google.common.primitives.Ints; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.slf4j.event.Level; |
| |
| /** Unit tests for IPC. */ |
| public class TestIPC { |
| public static final Logger LOG = LoggerFactory.getLogger(TestIPC.class); |
| |
| private static Configuration conf; |
| final static int PING_INTERVAL = 1000; |
| final static private int MIN_SLEEP_TIME = 1000; |
| /** |
| * Flag used to turn off the fault injection behavior |
| * of the various writables. |
| **/ |
| static boolean WRITABLE_FAULTS_ENABLED = true; |
| static int WRITABLE_FAULTS_SLEEP = 0; |
| |
| @Before |
| public void setupConf() { |
| conf = new Configuration(); |
| Client.setPingInterval(conf, PING_INTERVAL); |
| // tests may enable security, so disable before each test |
| UserGroupInformation.setConfiguration(conf); |
| } |
| |
| static final Random RANDOM = new Random(); |
| |
| private static final String ADDRESS = "0.0.0.0"; |
| |
| /** Directory where we can count open file descriptors on Linux */ |
| private static final File FD_DIR = new File("/proc/self/fd"); |
| |
| static ConnectionId getConnectionId(InetSocketAddress addr, int rpcTimeout, |
| Configuration conf) throws IOException { |
| return ConnectionId.getConnectionId(addr, null, |
| UserGroupInformation.getCurrentUser(), rpcTimeout, null, conf); |
| } |
| |
| static Writable call(Client client, InetSocketAddress addr, |
| int serviceClass, Configuration conf) throws IOException { |
| final LongWritable param = new LongWritable(RANDOM.nextLong()); |
| final ConnectionId remoteId = getConnectionId(addr, MIN_SLEEP_TIME, conf); |
| return client.call(RPC.RpcKind.RPC_BUILTIN, param, remoteId, serviceClass, |
| null); |
| } |
| |
| static LongWritable call(Client client, long param, InetSocketAddress addr, |
| Configuration conf) throws IOException { |
| return call(client, new LongWritable(param), addr, 0, conf); |
| } |
| |
| static LongWritable call(Client client, LongWritable param, |
| InetSocketAddress addr, int rpcTimeout, Configuration conf) |
| throws IOException { |
| final ConnectionId remoteId = getConnectionId(addr, rpcTimeout, conf); |
| return (LongWritable)client.call(RPC.RpcKind.RPC_BUILTIN, param, remoteId, |
| RPC.RPC_SERVICE_CLASS_DEFAULT, null); |
| } |
| |
| static class TestServer extends Server { |
| // Tests can set callListener to run a piece of code each time the server |
| // receives a call. This code executes on the server thread, so it has |
| // visibility of that thread's thread-local storage. |
| Runnable callListener; |
| private boolean sleep; |
| private Class<? extends Writable> responseClass; |
| |
| public TestServer(int handlerCount, boolean sleep) throws IOException { |
| this(handlerCount, sleep, LongWritable.class, null); |
| } |
| |
| public TestServer(int port, int handlerCount, boolean sleep) |
| throws IOException { |
| this(port, handlerCount, sleep, LongWritable.class, null); |
| } |
| |
| public TestServer(int handlerCount, boolean sleep, Configuration conf) |
| throws IOException { |
| this(handlerCount, sleep, LongWritable.class, null, conf); |
| } |
| |
| public TestServer(int handlerCount, boolean sleep, |
| Class<? extends Writable> paramClass, |
| Class<? extends Writable> responseClass) throws IOException { |
| this(handlerCount, sleep, paramClass, responseClass, conf); |
| } |
| |
| public TestServer(int port, int handlerCount, boolean sleep, |
| Class<? extends Writable> paramClass, |
| Class<? extends Writable> responseClass) throws IOException { |
| this(port, handlerCount, sleep, paramClass, responseClass, conf); |
| } |
| |
| public TestServer(int handlerCount, boolean sleep, |
| Class<? extends Writable> paramClass, |
| Class<? extends Writable> responseClass, Configuration conf) |
| throws IOException { |
| this(0, handlerCount, sleep, paramClass, responseClass, conf); |
| } |
| |
| public TestServer(int port, int handlerCount, boolean sleep, |
| Class<? extends Writable> paramClass, |
| Class<? extends Writable> responseClass, Configuration conf) |
| throws IOException { |
| super(ADDRESS, port, paramClass, handlerCount, conf); |
| this.sleep = sleep; |
| this.responseClass = responseClass; |
| } |
| |
| @Override |
| public Writable call(RPC.RpcKind rpcKind, String protocol, Writable param, |
| long receiveTime) throws IOException { |
| if (sleep) { |
| // sleep a bit |
| try { |
| Thread.sleep(RANDOM.nextInt(PING_INTERVAL) + MIN_SLEEP_TIME); |
| } catch (InterruptedException e) {} |
| } |
| if (callListener != null) { |
| callListener.run(); |
| } |
| if (responseClass != null) { |
| try { |
| return responseClass.newInstance(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } else { |
| return param; // echo param as result |
| } |
| } |
| } |
| |
| private static class SerialCaller extends Thread { |
| private Client client; |
| private InetSocketAddress server; |
| private int count; |
| private boolean failed; |
| |
| public SerialCaller(Client client, InetSocketAddress server, int count) { |
| this.client = client; |
| this.server = server; |
| this.count = count; |
| } |
| |
| @Override |
| public void run() { |
| for (int i = 0; i < count; i++) { |
| try { |
| final long param = RANDOM.nextLong(); |
| LongWritable value = call(client, param, server, conf); |
| if (value.get() != param) { |
| LOG.error("Call failed!"); |
| failed = true; |
| break; |
| } |
| } catch (Exception e) { |
| LOG.error("Caught: " + StringUtils.stringifyException(e)); |
| failed = true; |
| } |
| } |
| } |
| } |
| |
| /** |
| * A RpcInvocationHandler instance for test. Its invoke function uses the same |
| * {@link Client} instance, and will fail the first totalRetry times (by |
| * throwing an IOException). |
| */ |
| private static class TestInvocationHandler implements RpcInvocationHandler { |
| private static int retry = 0; |
| private final Client client; |
| private final Server server; |
| private final int total; |
| |
| TestInvocationHandler(Client client, Server server, int total) { |
| this.client = client; |
| this.server = server; |
| this.total = total; |
| } |
| |
| protected Object returnValue(Object value) throws Exception { |
| if (retry++ < total) { |
| throw new IOException("Fake IOException"); |
| } |
| return value; |
| } |
| |
| @Override |
| public Object invoke(Object proxy, Method method, Object[] args) |
| throws Throwable { |
| LongWritable value = call(client, RANDOM.nextLong(), |
| NetUtils.getConnectAddress(server), conf); |
| return returnValue(value); |
| } |
| |
| @Override |
| public void close() throws IOException {} |
| |
| @Override |
| public ConnectionId getConnectionId() { |
| return null; |
| } |
| } |
| |
| private static class TestInvalidTokenHandler extends TestInvocationHandler { |
| private int invocations = 0; |
| TestInvalidTokenHandler(Client client, Server server) { |
| super(client, server, 1); |
| } |
| |
| @Override |
| protected Object returnValue(Object value) throws Exception { |
| throw new InvalidToken("Invalid Token"); |
| } |
| |
| @Override |
| public Object invoke(Object proxy, Method method, Object[] args) |
| throws Throwable { |
| invocations++; |
| return super.invoke(proxy, method, args); |
| } |
| } |
| |
| @Test(timeout=60000) |
| public void testSerial() throws IOException, InterruptedException { |
| internalTestSerial(3, false, 2, 5, 100); |
| internalTestSerial(3, true, 2, 5, 10); |
| } |
| |
| public void internalTestSerial(int handlerCount, boolean handlerSleep, |
| int clientCount, int callerCount, int callCount) |
| throws IOException, InterruptedException { |
| Server server = new TestServer(handlerCount, handlerSleep); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| Client[] clients = new Client[clientCount]; |
| for (int i = 0; i < clientCount; i++) { |
| clients[i] = new Client(LongWritable.class, conf); |
| } |
| |
| SerialCaller[] callers = new SerialCaller[callerCount]; |
| for (int i = 0; i < callerCount; i++) { |
| callers[i] = new SerialCaller(clients[i%clientCount], addr, callCount); |
| callers[i].start(); |
| } |
| for (int i = 0; i < callerCount; i++) { |
| callers[i].join(); |
| assertFalse(callers[i].failed); |
| } |
| for (int i = 0; i < clientCount; i++) { |
| clients[i].stop(); |
| } |
| server.stop(); |
| } |
| |
| @Test |
| public void testAuxiliaryPorts() throws IOException, InterruptedException { |
| int defaultPort = 9000; |
| int[] auxiliaryPorts = {9001, 9002, 9003}; |
| final int handlerCount = 5; |
| final boolean handlerSleep = false; |
| Server server = new TestServer(defaultPort, handlerCount, handlerSleep); |
| for (int port : auxiliaryPorts) { |
| server.addAuxiliaryListener(port); |
| } |
| Set<InetSocketAddress> listenerAddrs = |
| server.getAuxiliaryListenerAddresses(); |
| Set<InetSocketAddress> addrs = new HashSet<>(); |
| for (InetSocketAddress addr : listenerAddrs) { |
| addrs.add(NetUtils.getConnectAddress(addr)); |
| } |
| server.start(); |
| |
| Client client = new Client(LongWritable.class, conf); |
| Set<SerialCaller> calls = new HashSet<>(); |
| for (InetSocketAddress addr : addrs) { |
| calls.add(new SerialCaller(client, addr, 100)); |
| } |
| for (SerialCaller caller : calls) { |
| caller.join(); |
| assertFalse(caller.failed); |
| } |
| client.stop(); |
| server.stop(); |
| } |
| |
| @Test(timeout=60000) |
| public void testStandAloneClient() throws IOException { |
| Client client = new Client(LongWritable.class, conf); |
| InetSocketAddress address = new InetSocketAddress("127.0.0.1", 10); |
| try { |
| call(client, RANDOM.nextLong(), address, conf); |
| fail("Expected an exception to have been thrown"); |
| } catch (IOException e) { |
| String message = e.getMessage(); |
| String addressText = address.getHostName() + ":" + address.getPort(); |
| assertTrue("Did not find "+addressText+" in "+message, |
| message.contains(addressText)); |
| Throwable cause=e.getCause(); |
| assertNotNull("No nested exception in "+e,cause); |
| String causeText=cause.getMessage(); |
| assertTrue("Did not find " + causeText + " in " + message, |
| message.contains(causeText)); |
| } finally { |
| client.stop(); |
| } |
| } |
| |
| static void maybeThrowIOE() throws IOException { |
| if (WRITABLE_FAULTS_ENABLED) { |
| maybeSleep(); |
| throw new IOException("Injected fault"); |
| } |
| } |
| |
| static void maybeThrowRTE() { |
| if (WRITABLE_FAULTS_ENABLED) { |
| maybeSleep(); |
| throw new RuntimeException("Injected fault"); |
| } |
| } |
| |
| private static void maybeSleep() { |
| if (WRITABLE_FAULTS_SLEEP > 0) { |
| try { |
| Thread.sleep(WRITABLE_FAULTS_SLEEP); |
| } catch (InterruptedException ie) { |
| } |
| } |
| } |
| |
| @SuppressWarnings("unused") |
| private static class IOEOnReadWritable extends LongWritable { |
| public IOEOnReadWritable() {} |
| |
| @Override |
| public void readFields(DataInput in) throws IOException { |
| super.readFields(in); |
| maybeThrowIOE(); |
| } |
| } |
| |
| @SuppressWarnings("unused") |
| private static class RTEOnReadWritable extends LongWritable { |
| public RTEOnReadWritable() {} |
| |
| @Override |
| public void readFields(DataInput in) throws IOException { |
| super.readFields(in); |
| maybeThrowRTE(); |
| } |
| } |
| |
| @SuppressWarnings("unused") |
| private static class IOEOnWriteWritable extends LongWritable { |
| public IOEOnWriteWritable() {} |
| |
| @Override |
| public void write(DataOutput out) throws IOException { |
| super.write(out); |
| maybeThrowIOE(); |
| } |
| } |
| |
| @SuppressWarnings("unused") |
| private static class RTEOnWriteWritable extends LongWritable { |
| public RTEOnWriteWritable() {} |
| |
| @Override |
| public void write(DataOutput out) throws IOException { |
| super.write(out); |
| maybeThrowRTE(); |
| } |
| } |
| |
| /** |
| * Generic test case for exceptions thrown at some point in the IPC |
| * process. |
| * |
| * @param clientParamClass - client writes this writable for parameter |
| * @param serverParamClass - server reads this writable for parameter |
| * @param serverResponseClass - server writes this writable for response |
| * @param clientResponseClass - client reads this writable for response |
| */ |
| private void doErrorTest( |
| Class<? extends LongWritable> clientParamClass, |
| Class<? extends LongWritable> serverParamClass, |
| Class<? extends LongWritable> serverResponseClass, |
| Class<? extends LongWritable> clientResponseClass) |
| throws IOException, InstantiationException, IllegalAccessException { |
| |
| // start server |
| Server server = new TestServer(1, false, |
| serverParamClass, serverResponseClass); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| // start client |
| WRITABLE_FAULTS_ENABLED = true; |
| Client client = new Client(clientResponseClass, conf); |
| try { |
| LongWritable param = clientParamClass.newInstance(); |
| |
| try { |
| call(client, param, addr, 0, conf); |
| fail("Expected an exception to have been thrown"); |
| } catch (Throwable t) { |
| assertExceptionContains(t, "Injected fault"); |
| } |
| |
| // Doing a second call with faults disabled should return fine -- |
| // ie the internal state of the client or server should not be broken |
| // by the failed call |
| WRITABLE_FAULTS_ENABLED = false; |
| call(client, param, addr, 0, conf); |
| |
| } finally { |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| @Test(timeout=60000) |
| public void testIOEOnClientWriteParam() throws Exception { |
| doErrorTest(IOEOnWriteWritable.class, |
| LongWritable.class, |
| LongWritable.class, |
| LongWritable.class); |
| } |
| |
| @Test(timeout=60000) |
| public void testRTEOnClientWriteParam() throws Exception { |
| doErrorTest(RTEOnWriteWritable.class, |
| LongWritable.class, |
| LongWritable.class, |
| LongWritable.class); |
| } |
| |
| @Test(timeout=60000) |
| public void testIOEOnServerReadParam() throws Exception { |
| doErrorTest(LongWritable.class, |
| IOEOnReadWritable.class, |
| LongWritable.class, |
| LongWritable.class); |
| } |
| |
| @Test(timeout=60000) |
| public void testRTEOnServerReadParam() throws Exception { |
| doErrorTest(LongWritable.class, |
| RTEOnReadWritable.class, |
| LongWritable.class, |
| LongWritable.class); |
| } |
| |
| |
| @Test(timeout=60000) |
| public void testIOEOnServerWriteResponse() throws Exception { |
| doErrorTest(LongWritable.class, |
| LongWritable.class, |
| IOEOnWriteWritable.class, |
| LongWritable.class); |
| } |
| |
| @Test(timeout=60000) |
| public void testRTEOnServerWriteResponse() throws Exception { |
| doErrorTest(LongWritable.class, |
| LongWritable.class, |
| RTEOnWriteWritable.class, |
| LongWritable.class); |
| } |
| |
| @Test(timeout=60000) |
| public void testIOEOnClientReadResponse() throws Exception { |
| doErrorTest(LongWritable.class, |
| LongWritable.class, |
| LongWritable.class, |
| IOEOnReadWritable.class); |
| } |
| |
| @Test(timeout=60000) |
| public void testRTEOnClientReadResponse() throws Exception { |
| doErrorTest(LongWritable.class, |
| LongWritable.class, |
| LongWritable.class, |
| RTEOnReadWritable.class); |
| } |
| |
| /** |
| * Test case that fails a write, but only after taking enough time |
| * that a ping should have been sent. This is a reproducer for a |
| * deadlock seen in one iteration of HADOOP-6762. |
| */ |
| @Test(timeout=60000) |
| public void testIOEOnWriteAfterPingClient() throws Exception { |
| // start server |
| Client.setPingInterval(conf, 100); |
| |
| try { |
| WRITABLE_FAULTS_SLEEP = 1000; |
| doErrorTest(IOEOnWriteWritable.class, |
| LongWritable.class, |
| LongWritable.class, |
| LongWritable.class); |
| } finally { |
| WRITABLE_FAULTS_SLEEP = 0; |
| } |
| } |
| |
| private static void assertExceptionContains( |
| Throwable t, String substring) { |
| String msg = StringUtils.stringifyException(t); |
| assertTrue("Exception should contain substring '" + substring + "':\n" + |
| msg, msg.contains(substring)); |
| LOG.info("Got expected exception", t); |
| } |
| |
| /** |
| * Test that, if the socket factory throws an IOE, it properly propagates |
| * to the client. |
| */ |
| @Test(timeout=60000) |
| public void testSocketFactoryException() throws IOException { |
| SocketFactory mockFactory = mock(SocketFactory.class); |
| doThrow(new IOException("Injected fault")).when(mockFactory).createSocket(); |
| Client client = new Client(LongWritable.class, conf, mockFactory); |
| |
| InetSocketAddress address = new InetSocketAddress("127.0.0.1", 10); |
| try { |
| call(client, RANDOM.nextLong(), address, conf); |
| fail("Expected an exception to have been thrown"); |
| } catch (IOException e) { |
| assertTrue(e.getMessage().contains("Injected fault")); |
| } finally { |
| client.stop(); |
| } |
| } |
| |
| /** |
| * Test that, if a RuntimeException is thrown after creating a socket |
| * but before successfully connecting to the IPC server, that the |
| * failure is handled properly. This is a regression test for |
| * HADOOP-7428. |
| */ |
| @Test(timeout=60000) |
| public void testRTEDuringConnectionSetup() throws IOException { |
| // Set up a socket factory which returns sockets which |
| // throw an RTE when setSoTimeout is called. |
| SocketFactory spyFactory = spy(NetUtils.getDefaultSocketFactory(conf)); |
| Mockito.doAnswer(new Answer<Socket>() { |
| @Override |
| public Socket answer(InvocationOnMock invocation) throws Throwable { |
| Socket s = spy((Socket)invocation.callRealMethod()); |
| doThrow(new RuntimeException("Injected fault")).when(s) |
| .setSoTimeout(anyInt()); |
| return s; |
| } |
| }).when(spyFactory).createSocket(); |
| |
| Server server = new TestServer(1, true); |
| Client client = new Client(LongWritable.class, conf, spyFactory); |
| server.start(); |
| try { |
| // Call should fail due to injected exception. |
| InetSocketAddress address = NetUtils.getConnectAddress(server); |
| try { |
| call(client, RANDOM.nextLong(), address, conf); |
| fail("Expected an exception to have been thrown"); |
| } catch (Exception e) { |
| LOG.info("caught expected exception", e); |
| assertTrue(StringUtils.stringifyException(e).contains( |
| "Injected fault")); |
| } |
| // Resetting to the normal socket behavior should succeed |
| // (i.e. it should not have cached a half-constructed connection) |
| |
| Mockito.reset(spyFactory); |
| call(client, RANDOM.nextLong(), address, conf); |
| } finally { |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| @Test(timeout=60000) |
| public void testIpcTimeout() throws IOException { |
| // start server |
| Server server = new TestServer(1, true); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| // start client |
| Client client = new Client(LongWritable.class, conf); |
| // set timeout to be less than MIN_SLEEP_TIME |
| try { |
| call(client, new LongWritable(RANDOM.nextLong()), addr, |
| MIN_SLEEP_TIME / 2, conf); |
| fail("Expected an exception to have been thrown"); |
| } catch (SocketTimeoutException e) { |
| LOG.info("Get a SocketTimeoutException ", e); |
| } |
| // set timeout to be bigger than 3*ping interval |
| call(client, new LongWritable(RANDOM.nextLong()), addr, |
| 3 * PING_INTERVAL + MIN_SLEEP_TIME, conf); |
| client.stop(); |
| } |
| |
| @Test(timeout=60000) |
| public void testIpcConnectTimeout() throws IOException { |
| // start server |
| Server server = new TestServer(1, true); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| //Intentionally do not start server to get a connection timeout |
| |
| // start client |
| Client.setConnectTimeout(conf, 100); |
| Client client = new Client(LongWritable.class, conf); |
| // set the rpc timeout to twice the MIN_SLEEP_TIME |
| try { |
| call(client, new LongWritable(RANDOM.nextLong()), addr, |
| MIN_SLEEP_TIME * 2, conf); |
| fail("Expected an exception to have been thrown"); |
| } catch (SocketTimeoutException e) { |
| LOG.info("Get a SocketTimeoutException ", e); |
| } |
| client.stop(); |
| } |
| |
| /** |
| * Check service class byte in IPC header is correct on wire. |
| */ |
| @Test(timeout=60000) |
| public void testIpcWithServiceClass() throws IOException { |
| // start server |
| Server server = new TestServer(5, false); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| // start client |
| Client.setConnectTimeout(conf, 10000); |
| |
| callAndVerify(server, addr, 0, true); |
| // Service Class is low to -128 as byte on wire. |
| // -128 shouldn't be casted on wire but -129 should. |
| callAndVerify(server, addr, -128, true); |
| callAndVerify(server, addr, -129, false); |
| |
| // Service Class is up to 127. |
| // 127 shouldn't be casted on wire but 128 should. |
| callAndVerify(server, addr, 127, true); |
| callAndVerify(server, addr, 128, false); |
| |
| server.stop(); |
| } |
| |
| private static class TestServerQueue extends Server { |
| final CountDownLatch firstCallLatch = new CountDownLatch(1); |
| final CountDownLatch callBlockLatch = new CountDownLatch(1); |
| |
| TestServerQueue(int expectedCalls, int readers, int callQ, int handlers, |
| Configuration conf) throws IOException { |
| super(ADDRESS, 0, LongWritable.class, handlers, readers, callQ, conf, null, null); |
| } |
| |
| @Override |
| public Writable call(RPC.RpcKind rpcKind, String protocol, Writable param, |
| long receiveTime) throws IOException { |
| firstCallLatch.countDown(); |
| try { |
| callBlockLatch.await(); |
| } catch (InterruptedException e) { |
| throw new IOException(e); |
| } |
| return param; |
| } |
| } |
| |
| /** |
| * Check that reader queueing works |
| * @throws BrokenBarrierException |
| * @throws InterruptedException |
| */ |
| @Test(timeout=60000) |
| public void testIpcWithReaderQueuing() throws Exception { |
| // 1 reader, 1 connectionQ slot, 1 callq |
| for (int i=0; i < 10; i++) { |
| checkBlocking(1, 1, 1); |
| } |
| // 4 readers, 5 connectionQ slots, 2 callq |
| for (int i=0; i < 10; i++) { |
| checkBlocking(4, 5, 2); |
| } |
| } |
| |
| // goal is to jam a handler with a connection, fill the callq with |
| // connections, in turn jamming the readers - then flood the server and |
| // ensure that the listener blocks when the reader connection queues fill |
| @SuppressWarnings("unchecked") |
| private void checkBlocking(int readers, int readerQ, int callQ) throws Exception { |
| int handlers = 1; // makes it easier |
| |
| final Configuration conf = new Configuration(); |
| conf.setInt(CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY, readerQ); |
| |
| // send in enough clients to block up the handlers, callq, and readers |
| final int initialClients = readers + callQ + handlers; |
| // max connections we should ever end up accepting at once |
| final int maxAccept = initialClients + readers*readerQ + 1; // 1 = listener |
| // stress it with 2X the max |
| int clients = maxAccept*2; |
| |
| final AtomicInteger failures = new AtomicInteger(0); |
| final CountDownLatch callFinishedLatch = new CountDownLatch(clients); |
| |
| // start server |
| final TestServerQueue server = |
| new TestServerQueue(clients, readers, callQ, handlers, conf); |
| CallQueueManager<Call> spy = spy( |
| (CallQueueManager<Call>)Whitebox.getInternalState(server, "callQueue")); |
| Whitebox.setInternalState(server, "callQueue", spy); |
| final InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| Client.setConnectTimeout(conf, 10000); |
| |
| // instantiate the threads, will start in batches |
| Thread[] threads = new Thread[clients]; |
| for (int i=0; i<clients; i++) { |
| threads[i] = new Thread(new Runnable() { |
| @Override |
| public void run() { |
| Client client = new Client(LongWritable.class, conf); |
| try { |
| call(client, new LongWritable(Thread.currentThread().getId()), |
| addr, 60000, conf); |
| } catch (Throwable e) { |
| LOG.error(e.toString()); |
| failures.incrementAndGet(); |
| return; |
| } finally { |
| callFinishedLatch.countDown(); |
| client.stop(); |
| } |
| } |
| }); |
| } |
| |
| // start enough clients to block up the handler, callq, and each reader; |
| // let the calls sequentially slot in to avoid some readers blocking |
| // and others not blocking in the race to fill the callq |
| for (int i=0; i < initialClients; i++) { |
| threads[i].start(); |
| if (i==0) { |
| // let first reader block in a call |
| server.firstCallLatch.await(); |
| } |
| // wait until reader put a call to callQueue, to make sure all readers |
| // are blocking on the queue after initialClients threads are started. |
| verify(spy, timeout(5000).times(i + 1)).put(Mockito.<Call>anyObject()); |
| } |
| |
| try { |
| // wait till everything is slotted, should happen immediately |
| GenericTestUtils.waitFor(new Supplier<Boolean>() { |
| @Override public Boolean get() { |
| return server.getNumOpenConnections() >= initialClients; |
| } |
| }, 100, 3000); |
| } catch (TimeoutException e) { |
| fail("timed out while waiting for connections to open."); |
| } |
| LOG.info("(initial clients) need:"+initialClients |
| +" connections have:"+server.getNumOpenConnections()); |
| LOG.info("ipc layer should be blocked"); |
| assertEquals(callQ, server.getCallQueueLen()); |
| assertEquals(initialClients, server.getNumOpenConnections()); |
| |
| // now flood the server with the rest of the connections, the reader's |
| // connection queues should fill and then the listener should block |
| for (int i=initialClients; i<clients; i++) { |
| threads[i].start(); |
| } |
| Thread.sleep(10); |
| |
| try { |
| GenericTestUtils.waitFor(new Supplier<Boolean>() { |
| @Override public Boolean get() { |
| return server.getNumOpenConnections() >= maxAccept; |
| } |
| }, 100, 3000); |
| } catch (TimeoutException e) { |
| fail("timed out while waiting for connections to open until maxAccept."); |
| } |
| LOG.info("(max clients) need:"+maxAccept |
| +" connections have:"+server.getNumOpenConnections()); |
| // check a few times to make sure we didn't go over |
| for (int i=0; i<4; i++) { |
| assertEquals(maxAccept, server.getNumOpenConnections()); |
| Thread.sleep(100); |
| } |
| |
| // sanity check that no calls have finished |
| assertEquals(clients, callFinishedLatch.getCount()); |
| LOG.info("releasing the calls"); |
| server.callBlockLatch.countDown(); |
| callFinishedLatch.await(); |
| for (Thread t : threads) { |
| t.join(); |
| } |
| assertEquals(0, failures.get()); |
| server.stop(); |
| } |
| |
| @Test(timeout=30000) |
| public void testConnectionIdleTimeouts() throws Exception { |
| GenericTestUtils.setLogLevel(Server.LOG, Level.DEBUG); |
| final int maxIdle = 1000; |
| final int cleanupInterval = maxIdle*3/4; // stagger cleanups |
| final int killMax = 3; |
| final int clients = 1 + killMax*2; // 1 to block, 2 batches to kill |
| |
| conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY, maxIdle); |
| conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_IDLETHRESHOLD_KEY, 0); |
| conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_KILL_MAX_KEY, killMax); |
| conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECTION_IDLESCANINTERVAL_KEY, cleanupInterval); |
| |
| final CyclicBarrier firstCallBarrier = new CyclicBarrier(2); |
| final CyclicBarrier callBarrier = new CyclicBarrier(clients); |
| final CountDownLatch allCallLatch = new CountDownLatch(clients); |
| final AtomicBoolean error = new AtomicBoolean(); |
| |
| final TestServer server = new TestServer(clients, false); |
| Thread[] threads = new Thread[clients]; |
| try { |
| server.callListener = new Runnable(){ |
| AtomicBoolean first = new AtomicBoolean(true); |
| @Override |
| public void run() { |
| try { |
| allCallLatch.countDown(); |
| // block first call |
| if (first.compareAndSet(true, false)) { |
| firstCallBarrier.await(); |
| } else { |
| callBarrier.await(); |
| } |
| } catch (Throwable t) { |
| LOG.error(t.toString()); |
| error.set(true); |
| } |
| } |
| }; |
| server.start(); |
| |
| // start client |
| final CountDownLatch callReturned = new CountDownLatch(clients-1); |
| final InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| final Configuration clientConf = new Configuration(); |
| clientConf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY, 10000); |
| for (int i=0; i < clients; i++) { |
| threads[i] = new Thread(new Runnable(){ |
| @Override |
| public void run() { |
| Client client = new Client(LongWritable.class, clientConf); |
| try { |
| call(client, Thread.currentThread().getId(), addr, clientConf); |
| callReturned.countDown(); |
| Thread.sleep(10000); |
| } catch (IOException e) { |
| LOG.error(e.toString()); |
| } catch (InterruptedException e) { |
| } finally { |
| client.stop(); |
| } |
| } |
| }); |
| threads[i].start(); |
| } |
| |
| // all calls blocked in handler so all connections made |
| allCallLatch.await(); |
| assertFalse(error.get()); |
| assertEquals(clients, server.getNumOpenConnections()); |
| |
| // wake up blocked calls and wait for client call to return, no |
| // connections should have closed |
| callBarrier.await(); |
| callReturned.await(); |
| assertEquals(clients, server.getNumOpenConnections()); |
| |
| // server won't close till maxIdle*2, so give scanning thread time to |
| // be almost ready to close idle connection. after which it should |
| // close max connections on every cleanupInterval |
| Thread.sleep(maxIdle*2-cleanupInterval); |
| for (int i=clients; i > 1; i -= killMax) { |
| Thread.sleep(cleanupInterval); |
| assertFalse(error.get()); |
| assertEquals(i, server.getNumOpenConnections()); |
| } |
| |
| // connection for the first blocked call should still be open |
| Thread.sleep(cleanupInterval); |
| assertFalse(error.get()); |
| assertEquals(1, server.getNumOpenConnections()); |
| |
| // wake up call and ensure connection times out |
| firstCallBarrier.await(); |
| Thread.sleep(maxIdle*2); |
| assertFalse(error.get()); |
| assertEquals(0, server.getNumOpenConnections()); |
| } finally { |
| for (Thread t : threads) { |
| if (t != null) { |
| t.interrupt(); |
| t.join(); |
| } |
| server.stop(); |
| } |
| } |
| } |
| |
| /** |
| * Make a call from a client and verify if header info is changed in server side |
| */ |
| private static void callAndVerify(Server server, InetSocketAddress addr, |
| int serviceClass, boolean noChanged) throws IOException{ |
| Client client = new Client(LongWritable.class, conf); |
| |
| call(client, addr, serviceClass, conf); |
| Connection connection = server.getConnections()[0]; |
| int serviceClass2 = connection.getServiceClass(); |
| assertFalse(noChanged ^ serviceClass == serviceClass2); |
| client.stop(); |
| } |
| |
| @Test(timeout=30000, expected=IOException.class) |
| public void testIpcAfterStopping() throws IOException { |
| // start server |
| Server server = new TestServer(5, false); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| // start client |
| Client client = new Client(LongWritable.class, conf); |
| call(client, addr, 0, conf); |
| client.stop(); |
| |
| // This call should throw IOException. |
| call(client, addr, 0, conf); |
| } |
| |
| /** |
| * Check that file descriptors aren't leaked by starting |
| * and stopping IPC servers. |
| */ |
| @Test(timeout=60000) |
| public void testSocketLeak() throws IOException { |
| Assume.assumeTrue(FD_DIR.exists()); // only run on Linux |
| |
| long startFds = countOpenFileDescriptors(); |
| for (int i = 0; i < 50; i++) { |
| Server server = new TestServer(1, true); |
| server.start(); |
| server.stop(); |
| } |
| long endFds = countOpenFileDescriptors(); |
| |
| assertTrue("Leaked " + (endFds - startFds) + " file descriptors", |
| endFds - startFds < 20); |
| } |
| |
| /** |
| * Check if Client is interrupted after handling |
| * InterruptedException during cleanup |
| */ |
| @Test(timeout=30000) |
| public void testInterrupted() { |
| Client client = new Client(LongWritable.class, conf); |
| Client.getClientExecutor().submit(new Runnable() { |
| public void run() { |
| while(true); |
| } |
| }); |
| Thread.currentThread().interrupt(); |
| client.stop(); |
| try { |
| assertTrue(Thread.currentThread().isInterrupted()); |
| LOG.info("Expected thread interrupt during client cleanup"); |
| } catch (AssertionError e) { |
| LOG.error("The Client did not interrupt after handling an Interrupted Exception"); |
| Assert.fail("The Client did not interrupt after handling an Interrupted Exception"); |
| } |
| // Clear Thread interrupt |
| Thread.interrupted(); |
| } |
| |
| private long countOpenFileDescriptors() { |
| return FD_DIR.list().length; |
| } |
| |
| @Test(timeout=60000) |
| public void testIpcFromHadoop_0_18_13() throws IOException { |
| doIpcVersionTest(NetworkTraces.HADOOP_0_18_3_RPC_DUMP, |
| NetworkTraces.RESPONSE_TO_HADOOP_0_18_3_RPC); |
| } |
| |
| @Test(timeout=60000) |
| public void testIpcFromHadoop0_20_3() throws IOException { |
| doIpcVersionTest(NetworkTraces.HADOOP_0_20_3_RPC_DUMP, |
| NetworkTraces.RESPONSE_TO_HADOOP_0_20_3_RPC); |
| } |
| |
| @Test(timeout=60000) |
| public void testIpcFromHadoop0_21_0() throws IOException { |
| doIpcVersionTest(NetworkTraces.HADOOP_0_21_0_RPC_DUMP, |
| NetworkTraces.RESPONSE_TO_HADOOP_0_21_0_RPC); |
| } |
| |
| @Test(timeout=60000) |
| public void testHttpGetResponse() throws IOException { |
| doIpcVersionTest("GET / HTTP/1.0\r\n\r\n".getBytes(), |
| Server.RECEIVED_HTTP_REQ_RESPONSE.getBytes()); |
| } |
| |
| @Test(timeout=60000) |
| public void testConnectionRetriesOnSocketTimeoutExceptions() throws IOException { |
| Configuration conf = new Configuration(); |
| // set max retries to 0 |
| conf.setInt( |
| CommonConfigurationKeysPublic.IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SOCKET_TIMEOUTS_KEY, |
| 0); |
| assertRetriesOnSocketTimeouts(conf, 1); |
| |
| // set max retries to 3 |
| conf.setInt( |
| CommonConfigurationKeysPublic.IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SOCKET_TIMEOUTS_KEY, |
| 3); |
| assertRetriesOnSocketTimeouts(conf, 4); |
| } |
| |
| static class CallInfo { |
| int id = RpcConstants.INVALID_CALL_ID; |
| int retry = RpcConstants.INVALID_RETRY_COUNT; |
| } |
| |
| /** |
| * Test if |
| * (1) the rpc server uses the call id/retry provided by the rpc client, and |
| * (2) the rpc client receives the same call id/retry from the rpc server. |
| */ |
| @Test(timeout=60000) |
| public void testCallIdAndRetry() throws IOException { |
| final CallInfo info = new CallInfo(); |
| |
| // Override client to store the call info and check response |
| final Client client = new Client(LongWritable.class, conf) { |
| @Override |
| Call createCall(RpcKind rpcKind, Writable rpcRequest) { |
| final Call call = super.createCall(rpcKind, rpcRequest); |
| info.id = call.id; |
| info.retry = call.retry; |
| return call; |
| } |
| |
| @Override |
| void checkResponse(RpcResponseHeaderProto header) throws IOException { |
| super.checkResponse(header); |
| Assert.assertEquals(info.id, header.getCallId()); |
| Assert.assertEquals(info.retry, header.getRetryCount()); |
| } |
| }; |
| |
| // Attach a listener that tracks every call received by the server. |
| final TestServer server = new TestServer(1, false); |
| server.callListener = new Runnable() { |
| @Override |
| public void run() { |
| Assert.assertEquals(info.id, Server.getCallId()); |
| Assert.assertEquals(info.retry, Server.getCallRetryCount()); |
| } |
| }; |
| |
| try { |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| final SerialCaller caller = new SerialCaller(client, addr, 10); |
| caller.run(); |
| assertFalse(caller.failed); |
| } finally { |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| /** A dummy protocol */ |
| interface DummyProtocol { |
| @Idempotent |
| public void dummyRun() throws IOException; |
| } |
| |
| /** |
| * Test the retry count while used in a retry proxy. |
| */ |
| @Test(timeout=60000) |
| public void testRetryProxy() throws IOException { |
| final Client client = new Client(LongWritable.class, conf); |
| |
| final TestServer server = new TestServer(1, false); |
| server.callListener = new Runnable() { |
| private int retryCount = 0; |
| @Override |
| public void run() { |
| Assert.assertEquals(retryCount++, Server.getCallRetryCount()); |
| } |
| }; |
| |
| // try more times, so it is easier to find race condition bug |
| // 10000 times runs about 6s on a core i7 machine |
| final int totalRetry = 10000; |
| DummyProtocol proxy = (DummyProtocol) Proxy.newProxyInstance( |
| DummyProtocol.class.getClassLoader(), |
| new Class[] { DummyProtocol.class }, new TestInvocationHandler(client, |
| server, totalRetry)); |
| DummyProtocol retryProxy = (DummyProtocol) RetryProxy.create( |
| DummyProtocol.class, proxy, RetryPolicies.RETRY_FOREVER); |
| |
| try { |
| server.start(); |
| retryProxy.dummyRun(); |
| Assert.assertEquals(TestInvocationHandler.retry, totalRetry + 1); |
| } finally { |
| Client.setCallIdAndRetryCount(0, 0, null); |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| /** |
| * Test that there is no retry when invalid token exception is thrown. |
| * Verfies fix for HADOOP-12054 |
| */ |
| @Test(expected = InvalidToken.class) |
| public void testNoRetryOnInvalidToken() throws IOException { |
| final Client client = new Client(LongWritable.class, conf); |
| final TestServer server = new TestServer(1, false); |
| TestInvalidTokenHandler handler = |
| new TestInvalidTokenHandler(client, server); |
| DummyProtocol proxy = (DummyProtocol) Proxy.newProxyInstance( |
| DummyProtocol.class.getClassLoader(), |
| new Class[] { DummyProtocol.class }, handler); |
| FailoverProxyProvider<DummyProtocol> provider = |
| new DefaultFailoverProxyProvider<DummyProtocol>( |
| DummyProtocol.class, proxy); |
| DummyProtocol retryProxy = |
| (DummyProtocol) RetryProxy.create(DummyProtocol.class, provider, |
| RetryPolicies.failoverOnNetworkException( |
| RetryPolicies.TRY_ONCE_THEN_FAIL, 100, 100, 10000, 0)); |
| |
| try { |
| server.start(); |
| retryProxy.dummyRun(); |
| } finally { |
| // Check if dummyRun called only once |
| Assert.assertEquals(handler.invocations, 1); |
| Client.setCallIdAndRetryCount(0, 0, null); |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| /** |
| * Test if the rpc server gets the default retry count (0) from client. |
| */ |
| @Test(timeout=60000) |
| public void testInitialCallRetryCount() throws IOException { |
| // Override client to store the call id |
| final Client client = new Client(LongWritable.class, conf); |
| |
| // Attach a listener that tracks every call ID received by the server. |
| final TestServer server = new TestServer(1, false); |
| server.callListener = new Runnable() { |
| @Override |
| public void run() { |
| // we have not set the retry count for the client, thus on the server |
| // side we should see retry count as 0 |
| Assert.assertEquals(0, Server.getCallRetryCount()); |
| } |
| }; |
| |
| try { |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| final SerialCaller caller = new SerialCaller(client, addr, 10); |
| caller.run(); |
| assertFalse(caller.failed); |
| } finally { |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| /** |
| * Test if the rpc server gets the retry count from client. |
| */ |
| @Test(timeout=60000) |
| public void testCallRetryCount() throws IOException { |
| final int retryCount = 255; |
| // Override client to store the call id |
| final Client client = new Client(LongWritable.class, conf); |
| Client.setCallIdAndRetryCount(Client.nextCallId(), 255, null); |
| |
| // Attach a listener that tracks every call ID received by the server. |
| final TestServer server = new TestServer(1, false); |
| server.callListener = new Runnable() { |
| @Override |
| public void run() { |
| // we have not set the retry count for the client, thus on the server |
| // side we should see retry count as 0 |
| Assert.assertEquals(retryCount, Server.getCallRetryCount()); |
| } |
| }; |
| |
| try { |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| final SerialCaller caller = new SerialCaller(client, addr, 10); |
| caller.run(); |
| assertFalse(caller.failed); |
| } finally { |
| client.stop(); |
| server.stop(); |
| } |
| } |
| |
| /** |
| * Tests that client generates a unique sequential call ID for each RPC call, |
| * even if multiple threads are using the same client. |
| * @throws InterruptedException |
| */ |
| @Test(timeout=60000) |
| public void testUniqueSequentialCallIds() |
| throws IOException, InterruptedException { |
| int serverThreads = 10, callerCount = 100, perCallerCallCount = 100; |
| TestServer server = new TestServer(serverThreads, false); |
| |
| // Attach a listener that tracks every call ID received by the server. This |
| // list must be synchronized, because multiple server threads will add to it. |
| final List<Integer> callIds = Collections.synchronizedList( |
| new ArrayList<Integer>()); |
| server.callListener = new Runnable() { |
| @Override |
| public void run() { |
| callIds.add(Server.getCallId()); |
| } |
| }; |
| |
| Client client = new Client(LongWritable.class, conf); |
| |
| try { |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| SerialCaller[] callers = new SerialCaller[callerCount]; |
| for (int i = 0; i < callerCount; ++i) { |
| callers[i] = new SerialCaller(client, addr, perCallerCallCount); |
| callers[i].start(); |
| } |
| for (int i = 0; i < callerCount; ++i) { |
| callers[i].join(); |
| assertFalse(callers[i].failed); |
| } |
| } finally { |
| client.stop(); |
| server.stop(); |
| } |
| |
| int expectedCallCount = callerCount * perCallerCallCount; |
| assertEquals(expectedCallCount, callIds.size()); |
| |
| // It is not guaranteed that the server executes requests in sequential order |
| // of client call ID, so we must sort the call IDs before checking that it |
| // contains every expected value. |
| Collections.sort(callIds); |
| final int startID = callIds.get(0).intValue(); |
| for (int i = 0; i < expectedCallCount; ++i) { |
| assertEquals(startID + i, callIds.get(i).intValue()); |
| } |
| } |
| |
| @Test |
| public void testMaxConnections() throws Exception { |
| conf.setInt("ipc.server.max.connections", 6); |
| Server server = null; |
| Thread connectors[] = new Thread[10]; |
| |
| try { |
| server = new TestServer(3, false); |
| final InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| assertEquals(0, server.getNumOpenConnections()); |
| |
| for (int i = 0; i < 10; i++) { |
| connectors[i] = new Thread() { |
| @Override |
| public void run() { |
| Socket sock = null; |
| try { |
| sock = NetUtils.getDefaultSocketFactory(conf).createSocket(); |
| NetUtils.connect(sock, addr, 3000); |
| try { |
| Thread.sleep(4000); |
| } catch (InterruptedException ie) { } |
| } catch (IOException ioe) { |
| } finally { |
| if (sock != null) { |
| try { |
| sock.close(); |
| } catch (IOException ioe) { } |
| } |
| } |
| } |
| }; |
| connectors[i].start(); |
| } |
| |
| Thread.sleep(1000); |
| // server should only accept up to 6 connections |
| assertEquals(6, server.getNumOpenConnections()); |
| // server should drop the other 4 connections |
| assertEquals(4, server.getNumDroppedConnections()); |
| |
| for (int i = 0; i < 10; i++) { |
| connectors[i].join(); |
| } |
| } finally { |
| if (server != null) { |
| server.stop(); |
| } |
| conf.setInt("ipc.server.max.connections", 0); |
| } |
| } |
| |
| @Test |
| public void testClientGetTimeout() throws IOException { |
| Configuration config = new Configuration(); |
| assertEquals(Client.getTimeout(config), -1); |
| } |
| |
| @Test(timeout=60000) |
| public void testSetupConnectionShouldNotBlockShutdown() throws Exception { |
| // Start server |
| SocketFactory mockFactory = Mockito.mock(SocketFactory.class); |
| Server server = new TestServer(1, true); |
| final InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| |
| // Track how many times we retried to set up the connection |
| final AtomicInteger createSocketCalled = new AtomicInteger(); |
| |
| doAnswer(new Answer() { |
| @Override |
| public Object answer(InvocationOnMock invocationOnMock) throws Throwable { |
| createSocketCalled.addAndGet(1); |
| Thread.sleep(MIN_SLEEP_TIME * 5); |
| throw new ConnectTimeoutException("fake"); |
| } |
| }).when(mockFactory).createSocket(); |
| final Client client = new Client(LongWritable.class, conf, mockFactory); |
| |
| final AtomicBoolean callStarted = new AtomicBoolean(false); |
| |
| // Call a random function asynchronously so that we can call stop() |
| new Thread(new Runnable() { |
| public void run() { |
| try { |
| callStarted.set(true); |
| call(client, RANDOM.nextLong(), addr, conf); |
| } catch (IOException ignored) {} |
| } |
| }).start(); |
| |
| GenericTestUtils.waitFor(new Supplier<Boolean>() { |
| @Override |
| public Boolean get() { |
| return callStarted.get() && createSocketCalled.get() == 1; |
| } |
| }, 50, 60000); |
| |
| // stop() should stop the client immediately without any more retries |
| client.stop(); |
| assertEquals(1, createSocketCalled.get()); |
| } |
| |
| private void assertRetriesOnSocketTimeouts(Configuration conf, |
| int maxTimeoutRetries) throws IOException { |
| SocketFactory mockFactory = Mockito.mock(SocketFactory.class); |
| doThrow(new ConnectTimeoutException("fake")).when(mockFactory).createSocket(); |
| Client client = new Client(LongWritable.class, conf, mockFactory); |
| InetSocketAddress address = new InetSocketAddress("127.0.0.1", 9090); |
| try { |
| call(client, RANDOM.nextLong(), address, conf); |
| fail("Not throwing the SocketTimeoutException"); |
| } catch (SocketTimeoutException e) { |
| Mockito.verify(mockFactory, Mockito.times(maxTimeoutRetries)) |
| .createSocket(); |
| } |
| client.stop(); |
| } |
| |
| @Test(timeout=4000) |
| public void testInsecureVersionMismatch() throws IOException { |
| checkVersionMismatch(); |
| } |
| |
| @Test(timeout=4000) |
| public void testSecureVersionMismatch() throws IOException { |
| SecurityUtil.setAuthenticationMethod(AuthenticationMethod.KERBEROS, conf); |
| UserGroupInformation.setConfiguration(conf); |
| checkVersionMismatch(); |
| } |
| |
| private void checkVersionMismatch() throws IOException { |
| try (final ServerSocket listenSocket = new ServerSocket()) { |
| listenSocket.bind(null); |
| InetSocketAddress addr = |
| (InetSocketAddress) listenSocket.getLocalSocketAddress(); |
| |
| // open a socket that accepts a client and immediately returns |
| // a version mismatch exception. |
| ExecutorService executor = Executors.newSingleThreadExecutor(); |
| executor.submit(new Runnable(){ |
| @Override |
| public void run() { |
| try { |
| Socket socket = listenSocket.accept(); |
| socket.getOutputStream().write( |
| NetworkTraces.RESPONSE_TO_HADOOP_0_20_3_RPC); |
| socket.close(); |
| } catch (Throwable t) { |
| // ignore. |
| } |
| } |
| }); |
| |
| try { |
| Client client = new Client(LongWritable.class, conf); |
| call(client, 0, addr, conf); |
| } catch (RemoteException re) { |
| Assert.assertEquals(RPC.VersionMismatch.class.getName(), |
| re.getClassName()); |
| Assert.assertEquals(NetworkTraces.HADOOP0_20_ERROR_MSG, |
| re.getMessage()); |
| return; |
| } |
| Assert.fail("didn't get version mismatch"); |
| } |
| } |
| |
| @Test |
| public void testRpcResponseLimit() throws Throwable { |
| Server server = new TestServer(1, false); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| |
| conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH, 0); |
| Client client = new Client(LongWritable.class, conf); |
| call(client, 0, addr, conf); |
| |
| conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH, 4); |
| client = new Client(LongWritable.class, conf); |
| try { |
| call(client, 0, addr, conf); |
| } catch (IOException ioe) { |
| Throwable t = ioe.getCause(); |
| Assert.assertNotNull(t); |
| Assert.assertEquals(RpcException.class, t.getClass()); |
| Assert.assertEquals("RPC response exceeds maximum data length", |
| t.getMessage()); |
| return; |
| } |
| Assert.fail("didn't get limit exceeded"); |
| } |
| |
| @Test |
| public void testUserBinding() throws Exception { |
| checkUserBinding(false); |
| } |
| |
| @Test |
| public void testProxyUserBinding() throws Exception { |
| checkUserBinding(true); |
| } |
| |
| private void checkUserBinding(boolean asProxy) throws Exception { |
| Socket s; |
| // don't attempt bind with no service host. |
| s = checkConnect(null, asProxy); |
| Mockito.verify(s, Mockito.never()).bind(Mockito.any(SocketAddress.class)); |
| |
| // don't attempt bind with service host not belonging to this host. |
| s = checkConnect("1.2.3.4", asProxy); |
| Mockito.verify(s, Mockito.never()).bind(Mockito.any(SocketAddress.class)); |
| |
| // do attempt bind when service host is this host. |
| InetAddress addr = InetAddress.getLocalHost(); |
| s = checkConnect(addr.getHostAddress(), asProxy); |
| Mockito.verify(s).bind(new InetSocketAddress(addr, 0)); |
| } |
| |
| // dummy protocol that claims to support kerberos. |
| @KerberosInfo(serverPrincipal = "server@REALM") |
| private static class TestBindingProtocol { |
| } |
| |
| private Socket checkConnect(String addr, boolean asProxy) throws Exception { |
| // create a fake ugi that claims to have kerberos credentials. |
| StringBuilder principal = new StringBuilder(); |
| principal.append("client"); |
| if (addr != null) { |
| principal.append("/").append(addr); |
| } |
| principal.append("@REALM"); |
| UserGroupInformation ugi = |
| spy(UserGroupInformation.createRemoteUser(principal.toString())); |
| Mockito.doReturn(true).when(ugi).hasKerberosCredentials(); |
| if (asProxy) { |
| ugi = UserGroupInformation.createProxyUser("proxy", ugi); |
| } |
| |
| // create a mock socket that throws on connect. |
| SocketException expectedConnectEx = |
| new SocketException("Expected connect failure"); |
| Socket s = Mockito.mock(Socket.class); |
| SocketFactory mockFactory = Mockito.mock(SocketFactory.class); |
| Mockito.doReturn(s).when(mockFactory).createSocket(); |
| doThrow(expectedConnectEx).when(s).connect( |
| Mockito.any(SocketAddress.class), Mockito.anyInt()); |
| |
| // do a dummy call and expect it to throw an exception on connect. |
| // tests should verify if/how a bind occurred. |
| try (Client client = new Client(LongWritable.class, conf, mockFactory)) { |
| final InetSocketAddress sockAddr = new InetSocketAddress(0); |
| final LongWritable param = new LongWritable(RANDOM.nextLong()); |
| final ConnectionId remoteId = new ConnectionId( |
| sockAddr, TestBindingProtocol.class, ugi, 0, |
| RetryPolicies.TRY_ONCE_THEN_FAIL, conf); |
| client.call(RPC.RpcKind.RPC_BUILTIN, param, remoteId, null); |
| fail("call didn't throw connect exception"); |
| } catch (SocketException se) { |
| // ipc layer re-wraps exceptions, so check the cause. |
| Assert.assertSame(expectedConnectEx, se.getCause()); |
| } |
| return s; |
| } |
| |
| private void doIpcVersionTest( |
| byte[] requestData, |
| byte[] expectedResponse) throws IOException { |
| Server server = new TestServer(1, true); |
| InetSocketAddress addr = NetUtils.getConnectAddress(server); |
| server.start(); |
| Socket socket = new Socket(); |
| |
| try { |
| NetUtils.connect(socket, addr, 5000); |
| |
| OutputStream out = socket.getOutputStream(); |
| InputStream in = socket.getInputStream(); |
| out.write(requestData, 0, requestData.length); |
| out.flush(); |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| IOUtils.copyBytes(in, baos, 256); |
| |
| byte[] responseData = baos.toByteArray(); |
| |
| assertEquals( |
| StringUtils.byteToHexString(expectedResponse), |
| StringUtils.byteToHexString(responseData)); |
| } finally { |
| IOUtils.closeSocket(socket); |
| server.stop(); |
| } |
| } |
| |
| /** |
| * Convert a string of lines that look like: |
| * "68 72 70 63 02 00 00 00 82 00 1d 6f 72 67 2e 61 hrpc.... ...org.a" |
| * .. into an array of bytes. |
| */ |
| private static byte[] hexDumpToBytes(String hexdump) { |
| final int LAST_HEX_COL = 3 * 16; |
| |
| StringBuilder hexString = new StringBuilder(); |
| |
| for (String line : StringUtils.toUpperCase(hexdump).split("\n")) { |
| hexString.append(line.substring(0, LAST_HEX_COL).replace(" ", "")); |
| } |
| return StringUtils.hexStringToByte(hexString.toString()); |
| } |
| |
| /** |
| * Wireshark traces collected from various client versions. These enable |
| * us to test that old versions of the IPC stack will receive the correct |
| * responses so that they will throw a meaningful error message back |
| * to the user. |
| */ |
| private static abstract class NetworkTraces { |
| /** |
| * Wireshark dump of an RPC request from Hadoop 0.18.3 |
| */ |
| final static byte[] HADOOP_0_18_3_RPC_DUMP = |
| hexDumpToBytes( |
| "68 72 70 63 02 00 00 00 82 00 1d 6f 72 67 2e 61 hrpc.... ...org.a\n" + |
| "70 61 63 68 65 2e 68 61 64 6f 6f 70 2e 69 6f 2e pache.ha doop.io.\n" + |
| "57 72 69 74 61 62 6c 65 00 30 6f 72 67 2e 61 70 Writable .0org.ap\n" + |
| "61 63 68 65 2e 68 61 64 6f 6f 70 2e 69 6f 2e 4f ache.had oop.io.O\n" + |
| "62 6a 65 63 74 57 72 69 74 61 62 6c 65 24 4e 75 bjectWri table$Nu\n" + |
| "6c 6c 49 6e 73 74 61 6e 63 65 00 2f 6f 72 67 2e llInstan ce./org.\n" + |
| "61 70 61 63 68 65 2e 68 61 64 6f 6f 70 2e 73 65 apache.h adoop.se\n" + |
| "63 75 72 69 74 79 2e 55 73 65 72 47 72 6f 75 70 curity.U serGroup\n" + |
| "49 6e 66 6f 72 6d 61 74 69 6f 6e 00 00 00 6c 00 Informat ion...l.\n" + |
| "00 00 00 00 12 67 65 74 50 72 6f 74 6f 63 6f 6c .....get Protocol\n" + |
| "56 65 72 73 69 6f 6e 00 00 00 02 00 10 6a 61 76 Version. .....jav\n" + |
| "61 2e 6c 61 6e 67 2e 53 74 72 69 6e 67 00 2e 6f a.lang.S tring..o\n" + |
| "72 67 2e 61 70 61 63 68 65 2e 68 61 64 6f 6f 70 rg.apach e.hadoop\n" + |
| "2e 6d 61 70 72 65 64 2e 4a 6f 62 53 75 62 6d 69 .mapred. JobSubmi\n" + |
| "73 73 69 6f 6e 50 72 6f 74 6f 63 6f 6c 00 04 6c ssionPro tocol..l\n" + |
| "6f 6e 67 00 00 00 00 00 00 00 0a ong..... ... \n"); |
| |
| final static String HADOOP0_18_ERROR_MSG = |
| "Server IPC version " + RpcConstants.CURRENT_VERSION + |
| " cannot communicate with client version 2"; |
| |
| /** |
| * Wireshark dump of the correct response that triggers an error message |
| * on an 0.18.3 client. |
| */ |
| final static byte[] RESPONSE_TO_HADOOP_0_18_3_RPC = |
| Bytes.concat(hexDumpToBytes( |
| "00 00 00 00 01 00 00 00 29 6f 72 67 2e 61 70 61 ........ )org.apa\n" + |
| "63 68 65 2e 68 61 64 6f 6f 70 2e 69 70 63 2e 52 che.hado op.ipc.R\n" + |
| "50 43 24 56 65 72 73 69 6f 6e 4d 69 73 6d 61 74 PC$Versi onMismat\n" + |
| "63 68 ch \n"), |
| Ints.toByteArray(HADOOP0_18_ERROR_MSG.length()), |
| HADOOP0_18_ERROR_MSG.getBytes()); |
| |
| /** |
| * Wireshark dump of an RPC request from Hadoop 0.20.3 |
| */ |
| final static byte[] HADOOP_0_20_3_RPC_DUMP = |
| hexDumpToBytes( |
| "68 72 70 63 03 00 00 00 79 27 6f 72 67 2e 61 70 hrpc.... y'org.ap\n" + |
| "61 63 68 65 2e 68 61 64 6f 6f 70 2e 69 70 63 2e ache.had oop.ipc.\n" + |
| "56 65 72 73 69 6f 6e 65 64 50 72 6f 74 6f 63 6f Versione dProtoco\n" + |
| "6c 01 0a 53 54 52 49 4e 47 5f 55 47 49 04 74 6f l..STRIN G_UGI.to\n" + |
| "64 64 09 04 74 6f 64 64 03 61 64 6d 07 64 69 61 dd..todd .adm.dia\n" + |
| "6c 6f 75 74 05 63 64 72 6f 6d 07 70 6c 75 67 64 lout.cdr om.plugd\n" + |
| "65 76 07 6c 70 61 64 6d 69 6e 05 61 64 6d 69 6e ev.lpadm in.admin\n" + |
| "0a 73 61 6d 62 61 73 68 61 72 65 06 6d 72 74 65 .sambash are.mrte\n" + |
| "73 74 00 00 00 6c 00 00 00 00 00 12 67 65 74 50 st...l.. ....getP\n" + |
| "72 6f 74 6f 63 6f 6c 56 65 72 73 69 6f 6e 00 00 rotocolV ersion..\n" + |
| "00 02 00 10 6a 61 76 61 2e 6c 61 6e 67 2e 53 74 ....java .lang.St\n" + |
| "72 69 6e 67 00 2e 6f 72 67 2e 61 70 61 63 68 65 ring..or g.apache\n" + |
| "2e 68 61 64 6f 6f 70 2e 6d 61 70 72 65 64 2e 4a .hadoop. mapred.J\n" + |
| "6f 62 53 75 62 6d 69 73 73 69 6f 6e 50 72 6f 74 obSubmis sionProt\n" + |
| "6f 63 6f 6c 00 04 6c 6f 6e 67 00 00 00 00 00 00 ocol..lo ng......\n" + |
| "00 14 .. \n"); |
| |
| final static String HADOOP0_20_ERROR_MSG = |
| "Server IPC version " + RpcConstants.CURRENT_VERSION + |
| " cannot communicate with client version 3"; |
| |
| |
| final static byte[] RESPONSE_TO_HADOOP_0_20_3_RPC = |
| Bytes.concat(hexDumpToBytes( |
| "ff ff ff ff ff ff ff ff 00 00 00 29 6f 72 67 2e ........ ...)org.\n" + |
| "61 70 61 63 68 65 2e 68 61 64 6f 6f 70 2e 69 70 apache.h adoop.ip\n" + |
| "63 2e 52 50 43 24 56 65 72 73 69 6f 6e 4d 69 73 c.RPC$Ve rsionMis\n" + |
| "6d 61 74 63 68 match \n"), |
| Ints.toByteArray(HADOOP0_20_ERROR_MSG.length()), |
| HADOOP0_20_ERROR_MSG.getBytes()); |
| |
| |
| final static String HADOOP0_21_ERROR_MSG = |
| "Server IPC version " + RpcConstants.CURRENT_VERSION + |
| " cannot communicate with client version 4"; |
| |
| final static byte[] HADOOP_0_21_0_RPC_DUMP = |
| hexDumpToBytes( |
| "68 72 70 63 04 50 hrpc.P" + |
| // in 0.21 it comes in two separate TCP packets |
| "00 00 00 3c 33 6f 72 67 2e 61 70 61 63 68 65 2e ...<3org .apache.\n" + |
| "68 61 64 6f 6f 70 2e 6d 61 70 72 65 64 75 63 65 hadoop.m apreduce\n" + |
| "2e 70 72 6f 74 6f 63 6f 6c 2e 43 6c 69 65 6e 74 .protoco l.Client\n" + |
| "50 72 6f 74 6f 63 6f 6c 01 00 04 74 6f 64 64 00 Protocol ...todd.\n" + |
| "00 00 00 71 00 00 00 00 00 12 67 65 74 50 72 6f ...q.... ..getPro\n" + |
| "74 6f 63 6f 6c 56 65 72 73 69 6f 6e 00 00 00 02 tocolVer sion....\n" + |
| "00 10 6a 61 76 61 2e 6c 61 6e 67 2e 53 74 72 69 ..java.l ang.Stri\n" + |
| "6e 67 00 33 6f 72 67 2e 61 70 61 63 68 65 2e 68 ng.3org. apache.h\n" + |
| "61 64 6f 6f 70 2e 6d 61 70 72 65 64 75 63 65 2e adoop.ma preduce.\n" + |
| "70 72 6f 74 6f 63 6f 6c 2e 43 6c 69 65 6e 74 50 protocol .ClientP\n" + |
| "72 6f 74 6f 63 6f 6c 00 04 6c 6f 6e 67 00 00 00 rotocol. .long...\n" + |
| "00 00 00 00 21 ....! \n"); |
| |
| final static byte[] RESPONSE_TO_HADOOP_0_21_0_RPC = |
| Bytes.concat(hexDumpToBytes( |
| "ff ff ff ff ff ff ff ff 00 00 00 29 6f 72 67 2e ........ ...)org.\n" + |
| "61 70 61 63 68 65 2e 68 61 64 6f 6f 70 2e 69 70 apache.h adoop.ip\n" + |
| "63 2e 52 50 43 24 56 65 72 73 69 6f 6e 4d 69 73 c.RPC$Ve rsionMis\n" + |
| "6d 61 74 63 68 match \n"), |
| Ints.toByteArray(HADOOP0_21_ERROR_MSG.length()), |
| HADOOP0_21_ERROR_MSG.getBytes()); |
| } |
| } |
