HADOOP-9786. RetryInvocationHandler#isRpcInvocation should support ProtocolTranslator. Contributed by Suresh Srinivas and Jing Zhao.

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

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -500,6 +500,9 @@ Release 2.1.0-beta - 2013-07-02
 
     HADOOP-9770. Make RetryCache#state non volatile. (suresh)
 
+    HADOOP-9786. RetryInvocationHandler#isRpcInvocation should support 
+    ProtocolTranslator. (suresh and jing9)
+
   OPTIMIZATIONS
 
     HADOOP-9150. Avoid unnecessary DNS resolution attempts for logical URIs

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/RetryInvocationHandler.java

@@ -30,12 +30,13 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.retry.RetryPolicy.RetryAction;
 import org.apache.hadoop.ipc.Client;
 import org.apache.hadoop.ipc.Client.ConnectionId;
+import org.apache.hadoop.ipc.ProtocolTranslator;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RpcConstants;
 import org.apache.hadoop.ipc.RpcInvocationHandler;
 import org.apache.hadoop.util.ThreadUtil;
 
-import com.google.common.base.Preconditions;
+import com.google.common.annotations.VisibleForTesting;
 
 class RetryInvocationHandler implements RpcInvocationHandler {
   public static final Log LOG = LogFactory.getLog(RetryInvocationHandler.class);
@@ -75,7 +76,7 @@ class RetryInvocationHandler implements RpcInvocationHandler {
     
     // The number of times this method invocation has been failed over.
     int invocationFailoverCount = 0;
-    final boolean isRpc = isRpcInvocation();
+    final boolean isRpc = isRpcInvocation(currentProxy);
     final int callId = isRpc? Client.nextCallId(): RpcConstants.INVALID_CALL_ID;
     int retries = 0;
     while (true) {
@@ -178,11 +179,15 @@ class RetryInvocationHandler implements RpcInvocationHandler {
     }
   }
 
-  private boolean isRpcInvocation() {
+  @VisibleForTesting
-    if (!Proxy.isProxyClass(currentProxy.getClass())) {
+  static boolean isRpcInvocation(Object proxy) {
+    if (proxy instanceof ProtocolTranslator) {
+      proxy = ((ProtocolTranslator) proxy).getUnderlyingProxyObject();
+    }
+    if (!Proxy.isProxyClass(proxy.getClass())) {
       return false;
     }
-    final InvocationHandler ih = Proxy.getInvocationHandler(currentProxy);
+    final InvocationHandler ih = Proxy.getInvocationHandler(proxy);
     return ih instanceof RpcInvocationHandler;
   }
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/retry/TestRetryProxy.java

@@ -35,6 +35,7 @@ import junit.framework.TestCase;
 
 import org.apache.hadoop.io.retry.UnreliableInterface.FatalException;
 import org.apache.hadoop.io.retry.UnreliableInterface.UnreliableException;
+import org.apache.hadoop.ipc.ProtocolTranslator;
 import org.apache.hadoop.ipc.RemoteException;
 
 public class TestRetryProxy extends TestCase {
@@ -58,6 +59,38 @@ public class TestRetryProxy extends TestCase {
     }
   }
   
+  /**
+   * Test for {@link RetryInvocationHandler#isRpcInvocation(Object)}
+   */
+  public void testRpcInvocation() throws Exception {
+    // For a proxy method should return true
+    final UnreliableInterface unreliable = (UnreliableInterface)
+      RetryProxy.create(UnreliableInterface.class, unreliableImpl, RETRY_FOREVER);
+    assertTrue(RetryInvocationHandler.isRpcInvocation(unreliable));
+    
+    // Embed the proxy in ProtocolTranslator
+    ProtocolTranslator xlator = new ProtocolTranslator() {
+      int count = 0;
+      @Override
+      public Object getUnderlyingProxyObject() {
+        count++;
+        return unreliable;
+      }
+      @Override
+      public String toString() {
+        return "" + count;
+      }
+    };
+    
+    // For a proxy wrapped in ProtocolTranslator method should return true
+    assertTrue(RetryInvocationHandler.isRpcInvocation(xlator));
+    // Ensure underlying proxy was looked at
+    assertEquals(xlator.toString(), "1");
+    
+    // For non-proxy the method must return false
+    assertFalse(RetryInvocationHandler.isRpcInvocation(new Object()));
+  }
+  
   public void testRetryForever() throws UnreliableException {
     UnreliableInterface unreliable = (UnreliableInterface)
       RetryProxy.create(UnreliableInterface.class, unreliableImpl, RETRY_FOREVER);
@@ -138,7 +171,7 @@ public class TestRetryProxy extends TestCase {
     }
   }
   
-  public void testRetryByRemoteException() throws UnreliableException {
+  public void testRetryByRemoteException() {
     Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap =
       Collections.<Class<? extends Exception>, RetryPolicy>singletonMap(FatalException.class, TRY_ONCE_THEN_FAIL);