HADOOP-1263. Change DFSClient to retry certain namenode calls with an exponential backoff. Contributed by Hairong.

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

CHANGES.txt

@@ -332,6 +332,10 @@ Trunk (unreleased changes)
 98. HADOOP-1184.  Fix HDFS decomissioning to complete when the only
     copy of a block is on a decommissioned node. (Dhruba Borthakur via cutting)
 
+99. HADOOP-1263.  Change DFSClient to retry certain namenode calls
+    with a random, exponentially increasing backoff time, to avoid
+    overloading the namenode on, e.g., job start.  (Hairong Kuang via cutting)
+
 
 Release 0.12.3 - 2007-04-06
 

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

@@ -18,6 +18,9 @@
 package org.apache.hadoop.dfs;
 
 import org.apache.hadoop.io.*;
+import org.apache.hadoop.io.retry.RetryPolicies;
+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.conf.*;
@@ -28,6 +31,7 @@ import org.apache.commons.logging.*;
 import java.io.*;
 import java.net.*;
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 
 /********************************************************
  * DFSClient can connect to a Hadoop Filesystem and 
@@ -94,6 +98,46 @@ class DFSClient implements FSConstants {
     Runtime.getRuntime().addShutdownHook(clientFinalizer);
   }
 
+  private static ClientProtocol createNamenode(
+      InetSocketAddress nameNodeAddr, Configuration conf)
+    throws IOException {
+    RetryPolicy timeoutPolicy = RetryPolicies.exponentialBackoffRetry(
+        5, 200, TimeUnit.MILLISECONDS);
+    RetryPolicy createPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
+        5, LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);
+    
+    Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
+      new HashMap<Class<? extends Exception>, RetryPolicy>();
+    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>();
+    
+    methodNameToPolicyMap.put("open", methodPolicy);
+    methodNameToPolicyMap.put("setReplication", methodPolicy);
+    methodNameToPolicyMap.put("abandonBlock", methodPolicy);
+    methodNameToPolicyMap.put("abandonFileInProgress", methodPolicy);
+    methodNameToPolicyMap.put("reportBadBlocks", methodPolicy);
+    methodNameToPolicyMap.put("exists", methodPolicy);
+    methodNameToPolicyMap.put("isDir", methodPolicy);
+    methodNameToPolicyMap.put("getListing", methodPolicy);
+    methodNameToPolicyMap.put("getHints", methodPolicy);
+    methodNameToPolicyMap.put("renewLease", methodPolicy);
+    methodNameToPolicyMap.put("getStats", methodPolicy);
+    methodNameToPolicyMap.put("getDatanodeReport", methodPolicy);
+    methodNameToPolicyMap.put("getBlockSize", methodPolicy);
+    methodNameToPolicyMap.put("getEditLogSize", methodPolicy);
+    methodNameToPolicyMap.put("complete", methodPolicy);
+    methodNameToPolicyMap.put("getEditLogSize", methodPolicy);
+    methodNameToPolicyMap.put("create", methodPolicy);
+
+    return (ClientProtocol) RetryProxy.create(ClientProtocol.class,
+        RPC.getProxy(ClientProtocol.class,
+            ClientProtocol.versionID, nameNodeAddr, conf),
+        methodNameToPolicyMap);
+  }
         
   /** 
    * Create a new DFSClient connected to the given namenode server.
@@ -101,8 +145,7 @@ class DFSClient implements FSConstants {
   public DFSClient(InetSocketAddress nameNodeAddr, Configuration conf)
     throws IOException {
     this.conf = conf;
-    this.namenode = (ClientProtocol) RPC.getProxy(ClientProtocol.class,
-                                                  ClientProtocol.versionID, nameNodeAddr, conf);
+    this.namenode = createNamenode(nameNodeAddr, conf);
     String taskId = conf.get("mapred.task.id");
     if (taskId != null) {
       this.clientName = "DFSClient_" + taskId; 
@@ -160,19 +203,12 @@ class DFSClient implements FSConstants {
   }
     
   public long getBlockSize(UTF8 f) throws IOException {
-    int retries = 4;
-    while (true) {
-      try {
-        return namenode.getBlockSize(f.toString());
-      } catch (IOException ie) {
-        if (--retries == 0) {
-          LOG.warn("Problem getting block size: " + 
-                   StringUtils.stringifyException(ie));
-          throw ie;
-        }
-        LOG.debug("Problem getting block size: " + 
-                  StringUtils.stringifyException(ie));
-      }
+    try {
+      return namenode.getBlockSize(f.toString());
+    } catch (IOException ie) {
+      LOG.warn("Problem getting block size: " + 
+          StringUtils.stringifyException(ie));
+      throw ie;
     }
   }
 
@@ -1133,31 +1169,8 @@ class DFSClient implements FSConstants {
     }
 
     private LocatedBlock locateNewBlock() throws IOException {     
-      int retries = 3;
-      while (true) {
-        while (true) {
-          try {
-            return namenode.create(src.toString(), clientName.toString(),
-                                   overwrite, replication, blockSize);
-          } catch (RemoteException e) {
-            if (--retries == 0 || 
-                !AlreadyBeingCreatedException.class.getName().
-                equals(e.getClassName())) {
-              throw e;
-            } else {
-              // because failed tasks take upto LEASE_PERIOD to
-              // release their pendingCreates files, if the file
-              // we want to create is already being created, 
-              // wait and try again.
-              LOG.info(StringUtils.stringifyException(e));
-              try {
-                Thread.sleep(LEASE_SOFTLIMIT_PERIOD);
-              } catch (InterruptedException ie) {
-              }
-            }
-          }
-        }
-      }
+      return namenode.create(src.toString(), clientName.toString(),
+          overwrite, replication, blockSize);
     }
         
     private LocatedBlock locateFollowingBlock(long start

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

@@ -67,7 +67,7 @@ class RetryInvocationHandler implements InvocationHandler {
           }
           return null;
         }
-        LOG.warn("Exception while invoking " + method.getName()
+        LOG.info("Exception while invoking " + method.getName()
                  + " of " + implementation.getClass() + ". Retrying."
                  + StringUtils.stringifyException(e));
       }
@@ -76,6 +76,9 @@ class RetryInvocationHandler implements InvocationHandler {
 
   private Object invokeMethod(Method method, Object[] args) throws Throwable {
     try {
+      if (!method.isAccessible()) {
+        method.setAccessible(true);
+      }
       return method.invoke(implementation, args);
     } catch (InvocationTargetException e) {
       throw e.getCause();

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.io.retry;
 
 import java.util.Map;
+import java.util.Random;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -82,6 +83,10 @@ public class RetryPolicies {
     return new RetryUpToMaximumCountWithProportionalSleep(maxRetries, sleepTime, timeUnit);
   }
   
+  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.
@@ -121,7 +126,7 @@ public class RetryPolicies {
     }
 
     public boolean shouldRetry(Exception e, int retries) throws Exception {
-      if (retries > maxRetries) {
+      if (retries >= maxRetries) {
         throw e;
       }
       try {
@@ -184,5 +189,16 @@ public class RetryPolicies {
     
   }
   
-  
+  static class ExponentialBackoffRetry extends RetryLimited {
+    private Random r = new Random();
+    public ExponentialBackoffRetry(
+        int maxRetries, long sleepTime, TimeUnit timeUnit) {
+      super(maxRetries, sleepTime, timeUnit);
+    }
+    
+    @Override
+    protected long calculateSleepTime(int retries) {
+      return sleepTime*r.nextInt(1<<(retries+1));
+    }
+  }
 }

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

@@ -7,6 +7,7 @@ import static org.apache.hadoop.io.retry.RetryPolicies.retryByException;
 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;
+import static org.apache.hadoop.io.retry.RetryPolicies.exponentialBackoffRetry;
 
 import java.util.Collections;
 import java.util.Map;
@@ -101,6 +102,20 @@ public class TestRetryProxy extends TestCase {
     }
   }
   
+  public void testExponentialRetry() throws UnreliableException {
+    UnreliableInterface unreliable = (UnreliableInterface)
+      RetryProxy.create(UnreliableInterface.class, unreliableImpl,
+                        exponentialBackoffRetry(5, 1L, TimeUnit.NANOSECONDS));
+    unreliable.alwaysSucceeds();
+    unreliable.failsOnceThenSucceeds();
+    try {
+      unreliable.failsTenTimesThenSucceeds();
+      fail("Should fail");
+    } catch (UnreliableException e) {
+      // expected
+    }
+  }
+  
   public void testRetryByException() throws UnreliableException {
     Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap =
       Collections.<Class<? extends Exception>, RetryPolicy>singletonMap(FatalException.class, TRY_ONCE_THEN_FAIL);