AVRO-6422. Make RPC backend plugable.

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

Author: cutting

CHANGES.txt

@@ -49,6 +49,9 @@ Trunk (unreleased changes)
     HADOOP-6346. Add support for specifying unpack pattern regex to
     RunJar.unJar. (Todd Lipcon via tomwhite)
 
+    HADOOP-6422. Make RPC backend plugable, protocol-by-protocol, to
+    ease evolution towards Avro.  (cutting)
+
   OPTIMIZATIONS
 
   BUG FIXES

src/java/org/apache/hadoop/ipc/AvroRpc.java renamed to src/java/org/apache/hadoop/ipc/AvroRpcEngine.java

@@ -28,6 +28,8 @@
 import javax.net.SocketFactory;
 import javax.security.auth.login.LoginException;
 
+import org.apache.commons.logging.*;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -41,9 +43,14 @@
  * does not give cross-language wire compatibility, since the Hadoop RPC wire
  * format is non-standard, but it does permit use of Avro's protocol versioning
  * features for inter-Java RPCs. */
-public class AvroRpc {
+class AvroRpcEngine implements RpcEngine {
+  private static final Log LOG = LogFactory.getLog(RPC.class);
+
   private static int VERSION = 0;
 
+  // the implementation we tunnel through
+  private static final RpcEngine ENGINE = new WritableRpcEngine();
+
   /** Tunnel an Avro RPC request and response through Hadoop's RPC. */
   private static interface TunnelProtocol extends VersionedProtocol {
     /** All Avro methods and responses go through this. */
@@ -91,8 +98,9 @@ public ClientTransceiver(InetSocketAddress addr,
                              UserGroupInformation ticket,
                              Configuration conf, SocketFactory factory)
       throws IOException {
-      this.tunnel = (TunnelProtocol)RPC.getProxy(TunnelProtocol.class, VERSION,
-                                                 addr, ticket, conf, factory);
+      this.tunnel =
+        (TunnelProtocol)ENGINE.getProxy(TunnelProtocol.class, VERSION,
+                                        addr, ticket, conf, factory);
       this.remote = addr;
     }
 
@@ -111,44 +119,48 @@ public void writeBuffers(List<ByteBuffer> buffers) throws IOException {
       throw new UnsupportedOperationException();
     }
 
-    public void close() throws IOException {}
+    public void close() throws IOException {
+      ENGINE.stopProxy(tunnel);
+    }
   }
-    
+
   /** Construct a client-side proxy object that implements the named protocol,
    * talking to a server at the named address. */
-  public static Object getProxy(Class<?> protocol,
-                                InetSocketAddress addr,
-                                Configuration conf)
+  public Object getProxy(Class protocol, long clientVersion,
+                         InetSocketAddress addr, UserGroupInformation ticket,
+                         Configuration conf, SocketFactory factory)
     throws IOException {
-    UserGroupInformation ugi = null;
+    return Proxy.newProxyInstance
+      (protocol.getClassLoader(),
+       new Class[] { protocol },
+       new Invoker(protocol, addr, ticket, conf, factory));
+  }
+
+  /** Stop this proxy. */
+  public void stopProxy(Object proxy) {
     try {
-      ugi = UserGroupInformation.login(conf);
-    } catch (LoginException le) {
-      throw new RuntimeException("Couldn't login!");
+      ((Invoker)Proxy.getInvocationHandler(proxy)).close();
+    } catch (IOException e) {
+      LOG.warn("Error while stopping "+proxy, e);
     }
-    return getProxy(protocol, addr, ugi, conf,
-                    NetUtils.getDefaultSocketFactory(conf));
   }
 
-  /** Construct a client-side proxy object that implements the named protocol,
-   * talking to a server at the named address. */
-  public static Object getProxy
-    (final Class<?> protocol, final InetSocketAddress addr,
-     final UserGroupInformation ticket,
-     final Configuration conf, final SocketFactory factory)
-    throws IOException {
-
-    return Proxy.newProxyInstance
-      (protocol.getClassLoader(), new Class[] { protocol },
-       new InvocationHandler() {
-         public Object invoke(Object proxy, Method method, Object[] args) 
-           throws Throwable {
-           return new ReflectRequestor
-             (protocol,
-              new ClientTransceiver(addr, ticket, conf, factory))
-             .invoke(proxy, method, args);
-         }
-       });
+  private static class Invoker implements InvocationHandler, Closeable {
+    private final ClientTransceiver tx;
+    private final ReflectRequestor requestor;
+    public Invoker(Class<?> protocol, InetSocketAddress addr,
+                   UserGroupInformation ticket, Configuration conf,
+                   SocketFactory factory) throws IOException {
+      this.tx = new ClientTransceiver(addr, ticket, conf, factory);
+      this.requestor = new ReflectRequestor(protocol, tx);
+    }
+    @Override public Object invoke(Object proxy, Method method, Object[] args) 
+      throws Throwable {
+      return requestor.invoke(proxy, method, args);
+    }
+    public void close() throws IOException {
+      tx.close();
+    }
   }
 
   /** An Avro RPC Responder that can process requests passed via Hadoop RPC. */
@@ -170,24 +182,20 @@ public BufferListWritable call(final BufferListWritable request)
     }
   }
 
-  /** Construct a server for a protocol implementation instance listening on a
-   * port and address. */
-  public static Server getServer(Object impl, String bindAddress, int port,
-                                 Configuration conf) 
-    throws IOException {
-    return RPC.getServer(new TunnelResponder(impl.getClass(), impl),
-                         bindAddress, port, conf);
-
+  public Object[] call(Method method, Object[][] params,
+                       InetSocketAddress[] addrs, UserGroupInformation ticket,
+                       Configuration conf) throws IOException {
+    throw new UnsupportedOperationException();
   }
 
   /** Construct a server for a protocol implementation instance listening on a
    * port and address. */
-  public static RPC.Server getServer(Object impl, String bindAddress, int port,
-                                     int numHandlers, boolean verbose,
-                                     Configuration conf) 
-    throws IOException {
-    return RPC.getServer(new TunnelResponder(impl.getClass(), impl),
-                         bindAddress, port, numHandlers, verbose, conf);
+  public RPC.Server getServer(Class iface, Object impl, String bindAddress,
+                              int port, int numHandlers, boolean verbose,
+                              Configuration conf) throws IOException {
+    return ENGINE.getServer(TunnelProtocol.class,
+                            new TunnelResponder(iface, impl),
+                            bindAddress, port, numHandlers, verbose, conf);
   }
 
 }