HADOOP-1822. Allow the specialization and configuration of socket factories. Provide a StandardSocketFactory, and a SocksSocketFactory to allow the use of SOCKS proxies. (taton).

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@581734 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -77,6 +77,10 @@ Trunk (unreleased changes)
     HADOOP-1963.  Add a FileSystem implementation for the Kosmos
     Filesystem (KFS).  (Sriram Rao via cutting)
 
+    HADOOP-1822.  Allow the specialization and configuration of socket
+    factories. Provide a StandardSocketFactory, and a SocksSocketFactory to
+    allow the use of SOCKS proxies. (taton).
+
   OPTIMIZATIONS
 
     HADOOP-1910.  Reduce the number of RPCs that DistributedFileSystem.create()

conf/hadoop-default.xml

@@ -954,4 +954,39 @@ creations/deletions), or "all".</description>
   </description>
 </property>
 
+<!-- Proxy Configuration -->
+
+<property>
+  <name>hadoop.rpc.socket.factory.class.default</name>
+  <value>org.apache.hadoop.net.StandardSocketFactory</value>
+  <description> Default SocketFactory to use. This parameter is expected to be
+    formatted as "package.FactoryClassName".
+  </description>
+</property>
+
+<property>
+  <name>hadoop.rpc.socket.factory.class.ClientProtocol</name>
+  <value></value>
+  <description> SocketFactory to use to connect to a DFS. If null or empty, use
+    hadoop.rpc.socket.class.default. This socket factory is also used by
+    DFSClient to create sockets to DataNodes.
+  </description>
+</property>
+
+<property>
+  <name>hadoop.rpc.socket.factory.class.JobSubmissionProtocol</name>
+  <value></value>
+  <description> SocketFactory to use to connect to a Map/Reduce master
+    (JobTracker). If null or empty, then use hadoop.rpc.socket.class.default.
+  </description>
+</property>
+
+<property>
+  <name>hadoop.socks.server</name>
+  <value></value>
+  <description> Address (host:port) of the SOCKS server to be used by the
+    SocksSocketFactory.
+  </description>
+</property>
+
 </configuration>

src/java/org/apache/hadoop/dfs/DFSClient.java

@@ -23,6 +23,7 @@ import org.apache.hadoop.io.retry.RetryPolicy;
 import org.apache.hadoop.io.retry.RetryProxy;
 import org.apache.hadoop.fs.*;
 import org.apache.hadoop.ipc.*;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.conf.*;
 import org.apache.hadoop.dfs.DistributedFileSystem.DiskStatus;
 import org.apache.hadoop.util.*;
@@ -36,6 +37,8 @@ import java.util.zip.CRC32;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.ConcurrentHashMap;
 
+import javax.net.SocketFactory;
+
 /********************************************************
  * DFSClient can connect to a Hadoop Filesystem and 
  * perform basic file tasks.  It uses the ClientProtocol
@@ -60,6 +63,7 @@ class DFSClient implements FSConstants {
   private long defaultBlockSize;
   private short defaultReplication;
   private LocalDirAllocator dirAllocator;
+  private SocketFactory socketFactory;
     
   /**
    * A map from name -> DFSOutputStream of files that are currently being
@@ -142,7 +146,8 @@ class DFSClient implements FSConstants {
 
     return (ClientProtocol) RetryProxy.create(ClientProtocol.class,
         RPC.getProxy(ClientProtocol.class,
-            ClientProtocol.versionID, nameNodeAddr, conf),
+            ClientProtocol.versionID, nameNodeAddr, conf,
+            NetUtils.getSocketFactory(conf, ClientProtocol.class)),
         methodNameToPolicyMap);
   }
         
@@ -152,6 +157,7 @@ class DFSClient implements FSConstants {
   public DFSClient(InetSocketAddress nameNodeAddr, Configuration conf)
     throws IOException {
     this.conf = conf;
+    this.socketFactory = NetUtils.getSocketFactory(conf, ClientProtocol.class);
     this.namenode = createNamenode(nameNodeAddr, conf);
     String taskId = conf.get("mapred.task.id");
     if (taskId != null) {
@@ -984,7 +990,7 @@ class DFSClient implements FSConstants {
         InetSocketAddress targetAddr = retval.addr;
 
         try {
-          s = new Socket();
+          s = socketFactory.createSocket();
           s.connect(targetAddr, READ_TIMEOUT);
           s.setSoTimeout(READ_TIMEOUT);
           Block blk = targetBlock.getBlock();
@@ -1161,7 +1167,7 @@ class DFSClient implements FSConstants {
         InetSocketAddress targetAddr = retval.addr;
             
         try {
-          dn = new Socket();
+          dn = socketFactory.createSocket();
           dn.connect(targetAddr, READ_TIMEOUT);
           dn.setSoTimeout(READ_TIMEOUT);
               
@@ -1490,7 +1496,7 @@ class DFSClient implements FSConstants {
         //
         InetSocketAddress target = DataNode.createSocketAddr(nodes[0].getName());
         try {
-          s = new Socket();
+          s = socketFactory.createSocket();
           s.connect(target, READ_TIMEOUT);
           s.setSoTimeout(replication * READ_TIMEOUT);
         } catch (IOException ie) {

src/java/org/apache/hadoop/dfs/SecondaryNameNode.java

@@ -24,6 +24,7 @@ import org.apache.hadoop.conf.*;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Daemon;
 import org.apache.hadoop.mapred.StatusHttpServer;
+import org.apache.hadoop.net.NetUtils;
 
 import java.io.*;
 import java.net.*;
@@ -93,8 +94,10 @@ public class SecondaryNameNode implements FSConstants, Runnable {
     nameNodeAddr = DataNode.createSocketAddr(
                                              conf.get("fs.default.name", "local"));
     this.conf = conf;
-    this.namenode = (ClientProtocol) RPC.getProxy(ClientProtocol.class,
-                                                  ClientProtocol.versionID, nameNodeAddr, conf);
+    this.namenode =
+        (ClientProtocol) RPC.getProxy(ClientProtocol.class,
+            ClientProtocol.versionID, nameNodeAddr, conf, NetUtils
+                .getSocketFactory(conf, ClientProtocol.class));
 
     //
     // initialize the webserver for uploading files.

src/java/org/apache/hadoop/ipc/Client.java

@@ -31,19 +31,20 @@ import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.FilterInputStream;
 import java.io.FilterOutputStream;
-import java.io.OutputStream;
 
 import java.util.Hashtable;
 import java.util.Iterator;
 
+import javax.net.SocketFactory;
+
 import org.apache.commons.logging.*;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.dfs.FSConstants;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
 import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 
@@ -71,6 +72,7 @@ public class Client {
   private int maxIdleTime; //connections will be culled if it was idle for 
                            //maxIdleTime msecs
   private int maxRetries; //the max. no. of retries for socket connections
+  private SocketFactory socketFactory;           // how to create sockets
 
   /** A call waiting for a value. */
   private class Call {
@@ -146,7 +148,7 @@ public class Client {
       short failures = 0;
       while (true) {
         try {
-          this.socket = new Socket();
+          this.socket = socketFactory.createSocket();
           this.socket.connect(address, FSConstants.READ_TIMEOUT);
           break;
         } catch (IOException ie) { //SocketTimeoutException is also caught 
@@ -426,13 +428,14 @@ public class Client {
 
   /** Construct an IPC client whose values are of the given {@link Writable}
    * class. */
-  public Client(Class valueClass, Configuration conf) {
+  public Client(Class valueClass, Configuration conf, 
+      SocketFactory factory) {
     this.valueClass = valueClass;
     this.timeout = conf.getInt("ipc.client.timeout", 10000);
     this.maxIdleTime = conf.getInt("ipc.client.connection.maxidletime", 1000);
     this.maxRetries = conf.getInt("ipc.client.connect.max.retries", 10);
     this.conf = conf;
-
+    this.socketFactory = factory;
     Thread t = new ConnectionCuller();
     t.setDaemon(true);
     t.setName(valueClass.getName() + " Connection Culler");
@@ -440,6 +443,15 @@ public class Client {
               "Connection culler maxidletime= " + maxIdleTime + "ms");
     t.start();
   }
+
+  /**
+   * Construct an IPC client with the default SocketFactory
+   * @param valueClass
+   * @param conf
+   */
+  public Client(Class<?> valueClass, Configuration conf) {
+    this(valueClass, conf, NetUtils.getDefaultSocketFactory(conf));
+  }
  
   /** Stop all threads related to this client.  No further calls may be made
    * using this client. */

src/java/org/apache/hadoop/ipc/RPC.java

@@ -28,10 +28,15 @@ import java.net.ConnectException;
 import java.net.InetSocketAddress;
 import java.net.SocketTimeoutException;
 import java.io.*;
+import java.util.Map;
+import java.util.HashMap;
+
+import javax.net.SocketFactory;
 
 import org.apache.commons.logging.*;
 
 import org.apache.hadoop.io.*;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.conf.*;
 
 /** A simple RPC mechanism.
@@ -126,37 +131,51 @@ public class RPC {
 
   }
 
-  private static Client CLIENT;
+  private static Map<SocketFactory, Client> CLIENTS =
+      new HashMap<SocketFactory, Client>();
 
-  private static synchronized Client getClient(Configuration conf) {
+  private static synchronized Client getClient(Configuration conf,
+      SocketFactory factory) {
     // Construct & cache client.  The configuration is only used for timeout,
     // and Clients have connection pools.  So we can either (a) lose some
     // connection pooling and leak sockets, or (b) use the same timeout for all
     // configurations.  Since the IPC is usually intended globally, not
     // per-job, we choose (a).
-    if (CLIENT == null) {
-      CLIENT = new Client(ObjectWritable.class, conf);
+    Client client = CLIENTS.get(factory);
+    if (client == null) {
+      client = new Client(ObjectWritable.class, conf, factory);
+      CLIENTS.put(factory, client);
     }
-    return CLIENT;
+    return client;
+  }
+  
+  /**
+   * Construct & cache client with the default SocketFactory.
+   * @param conf
+   * @return
+   */
+  private static Client getClient(Configuration conf) {
+    return getClient(conf, SocketFactory.getDefault());
   }
 
   /**
    * Stop all RPC client connections
    */
   public static synchronized void stopClient(){
-    if (CLIENT != null) {
-      CLIENT.stop();
-      CLIENT = null;
-    }
+    for (Client client : CLIENTS.values())
+      client.stop();
+    CLIENTS.clear();
   }
 
   private static class Invoker implements InvocationHandler {
     private InetSocketAddress address;
     private Client client;
 
-    public Invoker(InetSocketAddress address, Configuration conf) {
+    public Invoker(InetSocketAddress address, Configuration conf,
+        SocketFactory factory) {
+
       this.address = address;
-      this.client = getClient(conf);
+      this.client = getClient(conf, factory);
     }
 
     public Object invoke(Object proxy, Method method, Object[] args)
@@ -239,12 +258,14 @@ public class RPC {
   }
   /** Construct a client-side proxy object that implements the named protocol,
    * talking to a server at the named address. */
-  public static VersionedProtocol getProxy(Class protocol, long clientVersion,
-                                           InetSocketAddress addr, Configuration conf) throws IOException {
-    VersionedProtocol proxy = (VersionedProtocol) Proxy.newProxyInstance(
-                                                                         protocol.getClassLoader(),
-                                                                         new Class[] { protocol },
-                                                                         new Invoker(addr, conf));
+  public static VersionedProtocol getProxy(Class<?> protocol,
+      long clientVersion, InetSocketAddress addr, Configuration conf,
+      SocketFactory factory) throws IOException {
+
+    VersionedProtocol proxy =
+        (VersionedProtocol) Proxy.newProxyInstance(
+            protocol.getClassLoader(), new Class[] { protocol },
+            new Invoker(addr, conf, factory));
     long serverVersion = proxy.getProtocolVersion(protocol.getName(), 
                                                   clientVersion);
     if (serverVersion == clientVersion) {
@@ -255,6 +276,24 @@ public class RPC {
     }
   }
 
+  /**
+   * Construct a client-side proxy object with the default SocketFactory
+   * 
+   * @param protocol
+   * @param clientVersion
+   * @param addr
+   * @param conf
+   * @return a proxy instance
+   * @throws IOException
+   */
+  public static VersionedProtocol getProxy(Class<?> protocol,
+      long clientVersion, InetSocketAddress addr, Configuration conf)
+      throws IOException {
+
+    return getProxy(protocol, clientVersion, addr, conf, NetUtils
+        .getDefaultSocketFactory(conf));
+  }
+
   /** Expert: Make multiple, parallel calls to a set of servers. */
   public static Object[] call(Method method, Object[][] params,
                               InetSocketAddress[] addrs, Configuration conf)

src/java/org/apache/hadoop/mapred/JobClient.java

@@ -55,6 +55,7 @@ import org.apache.hadoop.io.retry.RetryPolicy;
 import org.apache.hadoop.io.retry.RetryProxy;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.mapred.TaskInProgress;
+import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
@@ -267,9 +268,10 @@ public class JobClient extends Configured implements MRConstants, Tool  {
   private JobSubmissionProtocol createProxy(InetSocketAddress addr,
                                             Configuration conf
                                             ) throws IOException {
-    JobSubmissionProtocol raw = (JobSubmissionProtocol) 
-      RPC.getProxy(JobSubmissionProtocol.class,
-                   JobSubmissionProtocol.versionID, addr, conf);
+    JobSubmissionProtocol raw =
+        (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
+            JobSubmissionProtocol.versionID, addr, conf, NetUtils
+                .getSocketFactory(conf, JobSubmissionProtocol.class));
     RetryPolicy backoffPolicy =
       RetryPolicies.retryUpToMaximumCountWithProportionalSleep
       (5, 10, java.util.concurrent.TimeUnit.SECONDS);

src/java/org/apache/hadoop/net/NetUtils.java

@@ -0,0 +1,78 @@
+package org.apache.hadoop.net;
+
+import javax.net.SocketFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.VersionedProtocol;
+import org.apache.hadoop.util.ReflectionUtils;
+
+public class NetUtils {
+
+  /**
+   * Get the socket factory for the given class according to its
+   * configuration parameter
+   * <tt>hadoop.rpc.socket.factory.class.&lt;ClassName&gt;</tt>. When no
+   * such parameter exists then fall back on the default socket factory as
+   * configured by <tt>hadoop.rpc.socket.factory.class.default</tt>. If
+   * this default socket factory is not configured, then fall back on the JVM
+   * default socket factory.
+   * 
+   * @param conf the configuration
+   * @param clazz the class (usually a {@link VersionedProtocol})
+   * @return a socket factory
+   */
+  public static SocketFactory getSocketFactory(Configuration conf,
+      Class<?> clazz) {
+
+    SocketFactory factory = null;
+
+    String propValue =
+        conf.get("hadoop.rpc.socket.factory.class." + clazz.getSimpleName());
+    if ((propValue != null) && (propValue.length() > 0))
+      factory = getSocketFactoryFromProperty(conf, propValue);
+
+    if (factory == null)
+      factory = getDefaultSocketFactory(conf);
+
+    return factory;
+  }
+
+  /**
+   * Get the default socket factory as specified by the configuration
+   * parameter <tt>hadoop.rpc.socket.factory.default</tt>
+   * 
+   * @param conf the configuration
+   * @return the default socket factory as specified in the configuration or
+   *         the JVM default socket factory if the configuration does not
+   *         contain a default socket factory property.
+   */
+  public static SocketFactory getDefaultSocketFactory(Configuration conf) {
+
+    String propValue = conf.get("hadoop.rpc.socket.factory.class.default");
+    if ((propValue == null) || (propValue.length() == 0))
+      return SocketFactory.getDefault();
+
+    return getSocketFactoryFromProperty(conf, propValue);
+  }
+
+  /**
+   * Get the socket factory corresponding to the given proxy URI. If the
+   * given proxy URI corresponds to an absence of configuration parameter,
+   * returns null. If the URI is malformed raises an exception.
+   * 
+   * @param propValue the property which is the class name of the
+   *        SocketFactory to instantiate; assumed non null and non empty.
+   * @return a socket factory as defined in the property value.
+   */
+  public static SocketFactory getSocketFactoryFromProperty(
+      Configuration conf, String propValue) {
+
+    try {
+      Class<?> theClass = conf.getClassByName(propValue);
+      return (SocketFactory) ReflectionUtils.newInstance(theClass, conf);
+
+    } catch (ClassNotFoundException cnfe) {
+      throw new RuntimeException("Socket Factory class not found: " + cnfe);
+    }
+  }
+}

src/java/org/apache/hadoop/net/SocksSocketFactory.java

@@ -0,0 +1,144 @@
+package org.apache.hadoop.net;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.net.Socket;
+import java.net.UnknownHostException;
+
+import javax.net.SocketFactory;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+
+/**
+ * Specialized SocketFactory to create sockets with a SOCKS proxy
+ */
+public class SocksSocketFactory extends SocketFactory implements
+    Configurable {
+
+  private Configuration conf;
+
+  private Proxy proxy;
+
+  /**
+   * Default empty constructor (for use with the reflection API).
+   */
+  public SocksSocketFactory() {
+    this.proxy = Proxy.NO_PROXY;
+  }
+
+  /**
+   * Constructor with a supplied Proxy
+   * 
+   * @param proxy the proxy to use to create sockets
+   */
+  public SocksSocketFactory(Proxy proxy) {
+    this.proxy = proxy;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket() throws IOException {
+
+    return new Socket(proxy);
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(InetAddress addr, int port) throws IOException {
+
+    Socket socket = createSocket();
+    socket.connect(new InetSocketAddress(addr, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(InetAddress addr, int port,
+      InetAddress localHostAddr, int localPort) throws IOException {
+
+    Socket socket = createSocket();
+    socket.bind(new InetSocketAddress(localHostAddr, localPort));
+    socket.connect(new InetSocketAddress(addr, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(String host, int port) throws IOException,
+      UnknownHostException {
+
+    Socket socket = createSocket();
+    socket.connect(new InetSocketAddress(host, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(String host, int port,
+      InetAddress localHostAddr, int localPort) throws IOException,
+      UnknownHostException {
+
+    Socket socket = createSocket();
+    socket.bind(new InetSocketAddress(localHostAddr, localPort));
+    socket.connect(new InetSocketAddress(host, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public int hashCode() {
+    return proxy.hashCode();
+  }
+
+  /* @inheritDoc */
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+    if (obj == null)
+      return false;
+    if (!(obj instanceof SocksSocketFactory))
+      return false;
+    final SocksSocketFactory other = (SocksSocketFactory) obj;
+    if (proxy == null) {
+      if (other.proxy != null)
+        return false;
+    } else if (!proxy.equals(other.proxy))
+      return false;
+    return true;
+  }
+
+  /* @inheritDoc */
+  public Configuration getConf() {
+    return this.conf;
+  }
+
+  /* @inheritDoc */
+  public void setConf(Configuration conf) {
+    this.conf = conf;
+    String proxyStr = conf.get("hadoop.socks.server");
+    if ((proxyStr != null) && (proxyStr.length() > 0)) {
+      setProxy(proxyStr);
+    }
+  }
+
+  /**
+   * Set the proxy of this socket factory as described in the string
+   * parameter
+   * 
+   * @param proxyStr the proxy address using the format "host:port"
+   */
+  private void setProxy(String proxyStr) {
+    String[] strs = proxyStr.split(":", 2);
+    if (strs.length != 2)
+      throw new RuntimeException("Bad SOCKS proxy parameter: " + proxyStr);
+    String host = strs[0];
+    int port = Integer.parseInt(strs[1]);
+    this.proxy =
+        new Proxy(Proxy.Type.SOCKS, InetSocketAddress.createUnresolved(host,
+            port));
+  }
+}

src/java/org/apache/hadoop/net/StandardSocketFactory.java

@@ -0,0 +1,89 @@
+package org.apache.hadoop.net;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.UnknownHostException;
+
+import javax.net.SocketFactory;
+
+/**
+ * Specialized SocketFactory to create sockets with a SOCKS proxy
+ */
+public class StandardSocketFactory extends SocketFactory {
+
+  /**
+   * Default empty constructor (for use with the reflection API).
+   */
+  public StandardSocketFactory() {
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket() throws IOException {
+    return new Socket();
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(InetAddress addr, int port) throws IOException {
+
+    Socket socket = createSocket();
+    socket.connect(new InetSocketAddress(addr, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(InetAddress addr, int port,
+      InetAddress localHostAddr, int localPort) throws IOException {
+
+    Socket socket = createSocket();
+    socket.bind(new InetSocketAddress(localHostAddr, localPort));
+    socket.connect(new InetSocketAddress(addr, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(String host, int port) throws IOException,
+      UnknownHostException {
+
+    Socket socket = createSocket();
+    socket.connect(new InetSocketAddress(host, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket(String host, int port,
+      InetAddress localHostAddr, int localPort) throws IOException,
+      UnknownHostException {
+
+    Socket socket = createSocket();
+    socket.bind(new InetSocketAddress(localHostAddr, localPort));
+    socket.connect(new InetSocketAddress(host, port));
+    return socket;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+    if (obj == null)
+      return false;
+    if (!(obj instanceof StandardSocketFactory))
+      return false;
+    return true;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public int hashCode() {
+    // Dummy hash code (to make find bugs happy)
+    return 47;
+  } 
+  
+}

src/test/org/apache/hadoop/ipc/TestSocketFactory.java

@@ -0,0 +1,189 @@
+/**
+ * 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 java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.dfs.DistributedFileSystem;
+import org.apache.hadoop.dfs.MiniDFSCluster;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.MiniMRCluster;
+import org.apache.hadoop.net.StandardSocketFactory;
+
+/**
+ * This class checks that RPCs can use specialized socket factories.
+ */
+public class TestSocketFactory extends TestCase {
+
+  /**
+   * Check that we can reach a NameNode or a JobTracker using a specific
+   * socket factory
+   */
+  public void testSocketFactory() throws IOException {
+    // Create a standard mini-cluster
+    Configuration sconf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(sconf, 1, true, null);
+    final int nameNodePort = cluster.getNameNodePort();
+
+    // Get a reference to its DFS directly
+    FileSystem fs = cluster.getFileSystem();
+    assertTrue(fs instanceof DistributedFileSystem);
+    DistributedFileSystem directDfs = (DistributedFileSystem) fs;
+
+    // Get another reference via network using a specific socket factory
+    Configuration cconf = new Configuration();
+    cconf.set("fs.default.name", String.format("hdfs://localhost:%s/",
+        nameNodePort + 10));
+    cconf.set("hadoop.rpc.socket.factory.class.default",
+        "org.apache.hadoop.ipc.DummySocketFactory");
+    cconf.set("hadoop.rpc.socket.factory.class.ClientProtocol",
+        "org.apache.hadoop.ipc.DummySocketFactory");
+    cconf.set("hadoop.rpc.socket.factory.class.JobSubmissionProtocol",
+        "org.apache.hadoop.ipc.DummySocketFactory");
+
+    fs = FileSystem.get(cconf);
+    assertTrue(fs instanceof DistributedFileSystem);
+    DistributedFileSystem dfs = (DistributedFileSystem) fs;
+
+    JobClient client = null;
+
+    try {
+      // This will test RPC to the NameNode only.
+      // could we test Client-DataNode connections?
+      Path filePath = new Path("/dir");
+
+      assertFalse(directDfs.exists(filePath));
+      assertFalse(dfs.exists(filePath));
+
+      directDfs.mkdirs(filePath);
+      assertTrue(directDfs.exists(filePath));
+      assertTrue(dfs.exists(filePath));
+
+      // This will test TPC to a JobTracker
+      MiniMRCluster mr = new MiniMRCluster(1, fs.getUri().toString(), 1);
+      final int jobTrackerPort = mr.getJobTrackerPort();
+
+      JobConf jconf = new JobConf(cconf);
+      jconf.set("mapred.job.tracker", String.format("localhost:%d",
+          jobTrackerPort + 10));
+      client = new JobClient(jconf);
+
+      JobStatus[] jobs = client.jobsToComplete();
+      assertTrue(jobs.length == 0);
+
+    } finally {
+      try {
+        if (client != null)
+          client.close();
+      } catch (Exception ignored) {
+        // nothing we can do
+        ignored.printStackTrace();
+      }
+      try {
+        if (dfs != null)
+          dfs.close();
+
+      } catch (Exception ignored) {
+        // nothing we can do
+        ignored.printStackTrace();
+      }
+      try {
+        if (directDfs != null)
+          directDfs.close();
+
+      } catch (Exception ignored) {
+        // nothing we can do
+        ignored.printStackTrace();
+      }
+      try {
+        if (cluster != null)
+          cluster.shutdown();
+
+      } catch (Exception ignored) {
+        // nothing we can do
+        ignored.printStackTrace();
+      }
+    }
+  }
+}
+
+/**
+ * Dummy socket factory which shift TPC ports by subtracting 10 when
+ * establishing a connection
+ */
+class DummySocketFactory extends StandardSocketFactory {
+  /**
+   * Default empty constructor (for use with the reflection API).
+   */
+  public DummySocketFactory() {
+  }
+
+  /* @inheritDoc */
+  @Override
+  public Socket createSocket() throws IOException {
+    return new Socket() {
+      @Override
+      public void connect(SocketAddress addr, int timeout)
+          throws IOException {
+
+        assert (addr instanceof InetSocketAddress);
+        InetSocketAddress iaddr = (InetSocketAddress) addr;
+        SocketAddress newAddr = null;
+        if (iaddr.isUnresolved())
+          newAddr =
+              new InetSocketAddress(iaddr.getHostName(),
+                  iaddr.getPort() - 10);
+        else
+          newAddr =
+              new InetSocketAddress(iaddr.getAddress(), iaddr.getPort() - 10);
+        System.out.printf("Test socket: rerouting %s to %s\n", iaddr,
+            newAddr);
+        super.connect(newAddr, timeout);
+      }
+    };
+  }
+
+  /* @inheritDoc */
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj)
+      return true;
+    if (obj == null)
+      return false;
+    if (!(obj instanceof DummySocketFactory))
+      return false;
+    return true;
+  }
+
+  /* @inheritDoc */
+  @Override
+  public int hashCode() {
+    // Dummy hash code (to make find bugs happy)
+    return 53;
+  }
+}