Merge -r 541782:541783 from trunk to 0.13 branch. Fixes: HADOOP-1411.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/branches/branch-0.13@541785 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -426,6 +426,10 @@ Branch 0.13 (unreleased changes)
      to indicate the default directory, per HADOOP-1386.
      (Hairong Kuang via cutting)
 
+128. HADOOP-1411.  Make task retry framework handle 
+     AlreadyBeingCreatedException when wrapped as a RemoteException.
+     (Hairong Kuang via tomwhite)
+
 
 Release 0.12.3 - 2007-04-06
 

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

@@ -106,11 +106,16 @@ class DFSClient implements FSConstants {
     RetryPolicy createPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
         5, LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);
     
+    Map<Class<? extends Exception>,RetryPolicy> remoteExceptionToPolicyMap =
+      new HashMap<Class<? extends Exception>, RetryPolicy>();
+    remoteExceptionToPolicyMap.put(AlreadyBeingCreatedException.class, createPolicy);
+
     Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
       new HashMap<Class<? extends Exception>, RetryPolicy>();
+    exceptionToPolicyMap.put(RemoteException.class, 
+        RetryPolicies.retryByRemoteException(
+            RetryPolicies.TRY_ONCE_THEN_FAIL, remoteExceptionToPolicyMap));
     exceptionToPolicyMap.put(SocketTimeoutException.class, timeoutPolicy);
-    exceptionToPolicyMap.put(AlreadyBeingCreatedException.class, createPolicy);
-
     RetryPolicy methodPolicy = RetryPolicies.retryByException(
         RetryPolicies.TRY_ONCE_THEN_FAIL, exceptionToPolicyMap);
     Map<String,RetryPolicy> methodNameToPolicyMap = new HashMap<String,RetryPolicy>();

src/java/org/apache/hadoop/io/retry/RetryInvocationHandler.java

@@ -59,7 +59,7 @@ class RetryInvocationHandler implements InvocationHandler {
         return invokeMethod(method, args);
       } catch (Exception e) {
         if (!policy.shouldRetry(e, retries++)) {
-          LOG.warn("Exception while invoking " + method.getName()
+          LOG.info("Exception while invoking " + method.getName()
                    + " of " + implementation.getClass() + ". Not retrying."
                    + StringUtils.stringifyException(e));
           if (!method.getReturnType().equals(Void.TYPE)) {
@@ -67,7 +67,7 @@ class RetryInvocationHandler implements InvocationHandler {
           }
           return null;
         }
-        LOG.info("Exception while invoking " + method.getName()
+        LOG.debug("Exception while invoking " + method.getName()
                  + " of " + implementation.getClass() + ". Retrying."
                  + StringUtils.stringifyException(e));
       }

src/java/org/apache/hadoop/io/retry/RetryPolicies.java

@@ -17,10 +17,15 @@
  */
 package org.apache.hadoop.io.retry;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Random;
+import java.util.Set;
+import java.util.Map.Entry;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.hadoop.ipc.RemoteException;
+
 /**
  * <p>
  * A collection of useful implementations of {@link RetryPolicy}.
@@ -83,10 +88,19 @@ public class RetryPolicies {
     return new RetryUpToMaximumCountWithProportionalSleep(maxRetries, sleepTime, timeUnit);
   }
   
+  /**
+   * <p>
+   * Keep trying a limited number of times, waiting a growing amount of time between attempts,
+   * and then fail by re-throwing the exception.
+   * The time between attempts is <code>sleepTime</code> mutliplied by a random
+   * number in the range of [0, 2 to the number of retries)
+   * </p>
+   */
   public static final RetryPolicy exponentialBackoffRetry(
       int maxRetries, long sleepTime, TimeUnit timeUnit) {
     return new ExponentialBackoffRetry(maxRetries, sleepTime, timeUnit);
   }
+  
   /**
    * <p>
    * Set a default policy with some explicit handlers for specific exceptions.
@@ -97,6 +111,18 @@ public class RetryPolicies {
     return new ExceptionDependentRetry(defaultPolicy, exceptionToPolicyMap);
   }
   
+  /**
+   * <p>
+   * A retry policy for RemoteException
+   * Set a default policy with some explicit handlers for specific exceptions.
+   * </p>
+   */
+  public static final RetryPolicy retryByRemoteException(
+      RetryPolicy defaultPolicy,
+      Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap) {
+    return new RemoteExceptionDependentRetry(defaultPolicy, exceptionToPolicyMap);
+  }
+  
   static class TryOnceThenFail implements RetryPolicy {
     public boolean shouldRetry(Exception e, int retries) throws Exception {
       throw e;
@@ -189,6 +215,35 @@ public class RetryPolicies {
     
   }
   
+  static class RemoteExceptionDependentRetry implements RetryPolicy {
+
+    RetryPolicy defaultPolicy;
+    Map<String, RetryPolicy> exceptionNameToPolicyMap;
+    
+    public RemoteExceptionDependentRetry(RetryPolicy defaultPolicy,
+                                   Map<Class<? extends Exception>,
+                                   RetryPolicy> exceptionToPolicyMap) {
+      this.defaultPolicy = defaultPolicy;
+      this.exceptionNameToPolicyMap = new HashMap<String, RetryPolicy>();
+      for (Entry<Class<? extends Exception>, RetryPolicy> e :
+          exceptionToPolicyMap.entrySet()) {
+        exceptionNameToPolicyMap.put(e.getKey().getName(), e.getValue());
+      }
+    }
+
+    public boolean shouldRetry(Exception e, int retries) throws Exception {
+      RetryPolicy policy = null;
+      if (e instanceof RemoteException) {
+        policy = exceptionNameToPolicyMap.get(
+            ((RemoteException) e).getClassName());
+      }
+      if (policy == null) {
+        policy = defaultPolicy;
+      }
+      return policy.shouldRetry(e, retries);
+    }
+  }
+  
   static class ExponentialBackoffRetry extends RetryLimited {
     private Random r = new Random();
     public ExponentialBackoffRetry(

src/test/org/apache/hadoop/io/retry/TestRetryProxy.java

@@ -4,6 +4,7 @@ import static org.apache.hadoop.io.retry.RetryPolicies.RETRY_FOREVER;
 import static org.apache.hadoop.io.retry.RetryPolicies.TRY_ONCE_DONT_FAIL;
 import static org.apache.hadoop.io.retry.RetryPolicies.TRY_ONCE_THEN_FAIL;
 import static org.apache.hadoop.io.retry.RetryPolicies.retryByException;
+import static org.apache.hadoop.io.retry.RetryPolicies.retryByRemoteException;
 import static org.apache.hadoop.io.retry.RetryPolicies.retryUpToMaximumCountWithFixedSleep;
 import static org.apache.hadoop.io.retry.RetryPolicies.retryUpToMaximumCountWithProportionalSleep;
 import static org.apache.hadoop.io.retry.RetryPolicies.retryUpToMaximumTimeWithFixedSleep;
@@ -17,6 +18,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.RemoteException;
 
 public class TestRetryProxy extends TestCase {
   
@@ -125,11 +127,26 @@ public class TestRetryProxy extends TestCase {
                         retryByException(RETRY_FOREVER, exceptionToPolicyMap));
     unreliable.failsOnceThenSucceeds();
     try {
-      unreliable.alwaysfailsWithFatalException();
+      unreliable.alwaysFailsWithFatalException();
       fail("Should fail");
     } catch (FatalException e) {
       // expected
     }
   }
   
+  public void testRetryByRemoteException() throws UnreliableException {
+    Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap =
+      Collections.<Class<? extends Exception>, RetryPolicy>singletonMap(FatalException.class, TRY_ONCE_THEN_FAIL);
+    
+    UnreliableInterface unreliable = (UnreliableInterface)
+      RetryProxy.create(UnreliableInterface.class, unreliableImpl,
+                        retryByRemoteException(RETRY_FOREVER, exceptionToPolicyMap));
+    try {
+      unreliable.alwaysFailsWithRemoteFatalException();
+      fail("Should fail");
+    } catch (RemoteException e) {
+      // expected
+    }
+  }  
+  
 }

src/test/org/apache/hadoop/io/retry/UnreliableImplementation.java

@@ -1,5 +1,7 @@
 package org.apache.hadoop.io.retry;
 
+import org.apache.hadoop.ipc.RemoteException;
+
 public class UnreliableImplementation implements UnreliableInterface {
 
   private int failsOnceInvocationCount,
@@ -10,9 +12,13 @@ public class UnreliableImplementation implements UnreliableInterface {
     // do nothing
   }
   
-  public void alwaysfailsWithFatalException() throws FatalException {
+  public void alwaysFailsWithFatalException() throws FatalException {
     throw new FatalException();
   }
+  
+  public void alwaysFailsWithRemoteFatalException() throws RemoteException {
+    throw new RemoteException(FatalException.class.getName(), "Oops");
+  }
 
   public void failsOnceThenSucceeds() throws UnreliableException {
     if (failsOnceInvocationCount++ == 0) {

src/test/org/apache/hadoop/io/retry/UnreliableInterface.java

@@ -1,5 +1,7 @@
 package org.apache.hadoop.io.retry;
 
+import org.apache.hadoop.ipc.RemoteException;
+
 public interface UnreliableInterface {
   
   public static class UnreliableException extends Exception {
@@ -12,7 +14,8 @@ public interface UnreliableInterface {
   
   void alwaysSucceeds() throws UnreliableException;
   
-  void alwaysfailsWithFatalException() throws FatalException;
+  void alwaysFailsWithFatalException() throws FatalException;
+  void alwaysFailsWithRemoteFatalException() throws RemoteException;
 
   void failsOnceThenSucceeds() throws UnreliableException;
   boolean failsOnceThenSucceedsWithReturnValue() throws UnreliableException;