accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [13/28] accumulo git commit: ACCUMULO-4317 Refactor TTimeoutTransport to be able to better test it
Date Tue, 21 Jun 2016 03:40:03 GMT
ACCUMULO-4317 Refactor TTimeoutTransport to be able to better test it


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/7bd9c088
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/7bd9c088
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/7bd9c088

Branch: refs/heads/master
Commit: 7bd9c0882efec7d42b83870b3cc3436471359472
Parents: bfc2a5b
Author: Josh Elser <elserj@apache.org>
Authored: Mon Jun 20 22:44:55 2016 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Mon Jun 20 22:44:55 2016 -0400

----------------------------------------------------------------------
 .../accumulo/core/rpc/TTimeoutTransport.java    | 146 ++++++++++++++++--
 .../core/rpc/TTimeoutTransportTest.java         | 151 +++++++++++++++++++
 2 files changed, 286 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/7bd9c088/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java b/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
index cc3f51b..809975f 100644
--- a/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
+++ b/core/src/main/java/org/apache/accumulo/core/rpc/TTimeoutTransport.java
@@ -30,16 +30,31 @@ import java.nio.channels.spi.SelectorProvider;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.thrift.transport.TIOStreamTransport;
 import org.apache.thrift.transport.TTransport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.common.net.HostAndPort;
 
+/**
+ * A utility class for setting up a {@link TTransport} with various necessary configurations
for ideal performance in Accumulo. These configurations include:
+ * <ul>
+ * <li>Setting SO_LINGER=false on the socket.</li>
+ * <li>Setting TCP_NO_DELAY=true on the socket.</li>
+ * <li>Setting timeouts on the I/OStreams.</li>
+ * </ul>
+ */
 public class TTimeoutTransport {
+  private static final Logger log = LoggerFactory.getLogger(TTimeoutTransport.class);
+
+  private static final TTimeoutTransport INSTANCE = new TTimeoutTransport();
+
+  private volatile Method GET_INPUT_STREAM_METHOD = null;
 
-  private static volatile Method GET_INPUT_STREAM_METHOD = null;
+  private TTimeoutTransport() {}
 
-  private static Method getNetUtilsInputStreamMethod() {
+  private Method getNetUtilsInputStreamMethod() {
     if (null == GET_INPUT_STREAM_METHOD) {
-      synchronized (TTimeoutTransport.class) {
+      synchronized (this) {
         if (null == GET_INPUT_STREAM_METHOD) {
           try {
             GET_INPUT_STREAM_METHOD = NetUtils.class.getMethod("getInputStream", Socket.class,
Long.TYPE);
@@ -53,35 +68,144 @@ public class TTimeoutTransport {
     return GET_INPUT_STREAM_METHOD;
   }
 
-  private static InputStream getInputStream(Socket socket, long timeout) {
+  /**
+   * Invokes the <code>NetUtils.getInputStream(Socket, long)</code> using reflection
to handle compatibility with both Hadoop 1 and 2.
+   *
+   * @param socket
+   *          The socket to create the input stream on
+   * @param timeout
+   *          The timeout for the input stream in milliseconds
+   * @return An InputStream on the socket
+   */
+  private InputStream getInputStream(Socket socket, long timeout) throws IOException {
     try {
       return (InputStream) getNetUtilsInputStreamMethod().invoke(null, socket, timeout);
     } catch (Exception e) {
-      throw new RuntimeException(e);
+      Throwable cause = e.getCause();
+      // Try to re-throw the IOException directly
+      if (cause instanceof IOException) {
+        throw (IOException) cause;
+      }
+
+      if (e instanceof RuntimeException) {
+        // Don't re-wrap another RTE around an RTE
+        throw (RuntimeException) e;
+      } else {
+        throw new RuntimeException(e);
+      }
     }
   }
 
+  /**
+   * Creates a Thrift TTransport to the given address with the given timeout. All created
resources are closed if an exception is thrown.
+   *
+   * @param addr
+   *          The address to connect the client to
+   * @param timeoutMillis
+   *          The timeout in milliseconds for the connection
+   * @return A TTransport connected to the given <code>addr</code>
+   * @throws IOException
+   *           If the transport fails to be created/connected
+   */
   public static TTransport create(HostAndPort addr, long timeoutMillis) throws IOException
{
-    return create(new InetSocketAddress(addr.getHostText(), addr.getPort()), timeoutMillis);
+    return INSTANCE.createInternal(new InetSocketAddress(addr.getHostText(), addr.getPort()),
timeoutMillis);
   }
 
+  /**
+   * Creates a Thrift TTransport to the given address with the given timeout. All created
resources are closed if an exception is thrown.
+   *
+   * @param addr
+   *          The address to connect the client to
+   * @param timeoutMillis
+   *          The timeout in milliseconds for the connection
+   * @return A TTransport connected to the given <code>addr</code>
+   * @throws IOException
+   *           If the transport fails to be created/connected
+   */
   public static TTransport create(SocketAddress addr, long timeoutMillis) throws IOException
{
+    return INSTANCE.createInternal(addr, timeoutMillis);
+  }
+
+  /**
+   * Opens a socket to the given <code>addr</code>, configures the socket, and
then creates a Thrift transport using the socket.
+   *
+   * @param addr
+   *          The address the socket should connect
+   * @param timeoutMillis
+   *          The socket timeout in milliseconds
+   * @return A TTransport instance to the given <code>addr</code>
+   * @throws IOException
+   *           If the Thrift client is failed to be connected/created
+   */
+  protected TTransport createInternal(SocketAddress addr, long timeoutMillis) throws IOException
{
     Socket socket = null;
     try {
-      socket = SelectorProvider.provider().openSocketChannel().socket();
+      socket = openSocket(addr);
+    } catch (IOException e) {
+      // openSocket handles closing the Socket on error
+      throw e;
+    }
+
+    // Should be non-null
+    assert null != socket;
+
+    // Set up the streams
+    try {
+      InputStream input = wrapInputStream(socket, timeoutMillis);
+      OutputStream output = wrapOutputStream(socket, timeoutMillis);
+      return new TIOStreamTransport(input, output);
+    } catch (IOException e) {
+      try {
+        socket.close();
+      } catch (IOException ioe) {
+        log.error("Failed to close socket after unsuccessful I/O stream setup", e);
+      }
+
+      throw e;
+    }
+  }
+
+  // Visible for testing
+  protected InputStream wrapInputStream(Socket socket, long timeoutMillis) throws IOException
{
+    return new BufferedInputStream(getInputStream(socket, timeoutMillis), 1024 * 10);
+  }
+
+  // Visible for testing
+  protected OutputStream wrapOutputStream(Socket socket, long timeoutMillis) throws IOException
{
+    return new BufferedOutputStream(NetUtils.getOutputStream(socket, timeoutMillis), 1024
* 10);
+  }
+
+  /**
+   * Opens and configures a {@link Socket} for Accumulo RPC.
+   *
+   * @param addr
+   *          The address to connect the socket to
+   * @return A socket connected to the given address, or null if the socket fails to connect
+   */
+  protected Socket openSocket(SocketAddress addr) throws IOException {
+    Socket socket = null;
+    try {
+      socket = openSocketChannel();
       socket.setSoLinger(false, 0);
       socket.setTcpNoDelay(true);
       socket.connect(addr);
-      InputStream input = new BufferedInputStream(getInputStream(socket, timeoutMillis),
1024 * 10);
-      OutputStream output = new BufferedOutputStream(NetUtils.getOutputStream(socket, timeoutMillis),
1024 * 10);
-      return new TIOStreamTransport(input, output);
+      return socket;
     } catch (IOException e) {
       try {
         if (socket != null)
           socket.close();
-      } catch (IOException ioe) {}
+      } catch (IOException ioe) {
+        log.error("Failed to close socket after unsuccessful open.", e);
+      }
 
       throw e;
     }
   }
+
+  /**
+   * Opens a socket channel and returns the underlying socket.
+   */
+  protected Socket openSocketChannel() throws IOException {
+    return SelectorProvider.provider().openSocketChannel().socket();
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/7bd9c088/core/src/test/java/org/apache/accumulo/core/rpc/TTimeoutTransportTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/rpc/TTimeoutTransportTest.java b/core/src/test/java/org/apache/accumulo/core/rpc/TTimeoutTransportTest.java
new file mode 100644
index 0000000..cedac9c
--- /dev/null
+++ b/core/src/test/java/org/apache/accumulo/core/rpc/TTimeoutTransportTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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.accumulo.core.rpc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.Socket;
+import java.net.SocketAddress;
+
+import org.junit.Test;
+
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.createMockBuilder;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for {@link TTimeoutTransport}.
+ */
+public class TTimeoutTransportTest {
+
+  void expectedSocketSetup(Socket s) throws IOException {
+    s.setSoLinger(false, 0);
+    expectLastCall().once();
+    s.setTcpNoDelay(true);
+    expectLastCall().once();
+  }
+
+  @Test
+  public void testFailedSocketOpenIsClosed() throws IOException {
+    SocketAddress addr = createMock(SocketAddress.class);
+    Socket s = createMock(Socket.class);
+    TTimeoutTransport timeoutTransport = createMockBuilder(TTimeoutTransport.class).addMockedMethod("openSocketChannel").createMock();
+
+    // Return out mocked socket
+    expect(timeoutTransport.openSocketChannel()).andReturn(s).once();
+
+    // tcpnodelay and solinger
+    expectedSocketSetup(s);
+
+    // Connect to the addr
+    s.connect(addr);
+    expectLastCall().andThrow(new IOException());
+
+    // The socket should be closed after the above IOException
+    s.close();
+
+    replay(addr, s, timeoutTransport);
+
+    try {
+      timeoutTransport.openSocket(addr);
+      fail("Expected to catch IOException but got none");
+    } catch (IOException e) {
+      // Expected
+    }
+
+    verify(addr, s, timeoutTransport);
+  }
+
+  @Test
+  public void testFailedInputStreamClosesSocket() throws IOException {
+    long timeout = 2 * 60 * 1000; // 2 mins
+    SocketAddress addr = createMock(SocketAddress.class);
+    Socket s = createMock(Socket.class);
+    TTimeoutTransport timeoutTransport = createMockBuilder(TTimeoutTransport.class).addMockedMethod("openSocketChannel").addMockedMethod("wrapInputStream")
+        .createMock();
+
+    // Return out mocked socket
+    expect(timeoutTransport.openSocketChannel()).andReturn(s).once();
+
+    // tcpnodelay and solinger
+    expectedSocketSetup(s);
+
+    // Connect to the addr
+    s.connect(addr);
+    expectLastCall().once();
+
+    expect(timeoutTransport.wrapInputStream(s, timeout)).andThrow(new IOException());
+
+    // The socket should be closed after the above IOException
+    s.close();
+
+    replay(addr, s, timeoutTransport);
+
+    try {
+      timeoutTransport.createInternal(addr, timeout);
+      fail("Expected to catch IOException but got none");
+    } catch (IOException e) {
+      // Expected
+    }
+
+    verify(addr, s, timeoutTransport);
+  }
+
+  @Test
+  public void testFailedOutputStreamClosesSocket() throws IOException {
+    long timeout = 2 * 60 * 1000; // 2 mins
+    SocketAddress addr = createMock(SocketAddress.class);
+    Socket s = createMock(Socket.class);
+    InputStream is = createMock(InputStream.class);
+    TTimeoutTransport timeoutTransport = createMockBuilder(TTimeoutTransport.class).addMockedMethod("openSocketChannel").addMockedMethod("wrapInputStream")
+        .addMockedMethod("wrapOutputStream").createMock();
+
+    // Return out mocked socket
+    expect(timeoutTransport.openSocketChannel()).andReturn(s).once();
+
+    // tcpnodelay and solinger
+    expectedSocketSetup(s);
+
+    // Connect to the addr
+    s.connect(addr);
+    expectLastCall().once();
+
+    // Input stream is set up
+    expect(timeoutTransport.wrapInputStream(s, timeout)).andReturn(is);
+    // Output stream fails to be set up
+    expect(timeoutTransport.wrapOutputStream(s, timeout)).andThrow(new IOException());
+
+    // The socket should be closed after the above IOException
+    s.close();
+
+    replay(addr, s, timeoutTransport);
+
+    try {
+      timeoutTransport.createInternal(addr, timeout);
+      fail("Expected to catch IOException but got none");
+    } catch (IOException e) {
+      // Expected
+    }
+
+    verify(addr, s, timeoutTransport);
+  }
+
+}


Mime
View raw message