diff --git a/CHANGES.txt b/CHANGES.txt
index c02fbd2042..f2c4ad4430 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -140,6 +140,10 @@ Trunk (unreleased changes)
     HADOOP-6534. Trim whitespace from directory lists initializing
     LocalDirAllocator. (Todd Lipcon via cdouglas)
 
+    HADOOP-6559. Makes the RPC client automatically re-login when the SASL 
+    connection setup fails. This is applicable only to keytab based logins.
+    (Devaraj Das)
+
   OPTIMIZATIONS
 
   BUG FIXES
diff --git a/src/java/org/apache/hadoop/ipc/Client.java b/src/java/org/apache/hadoop/ipc/Client.java
index f58f715b3d..c68cc7b290 100644
--- a/src/java/org/apache/hadoop/ipc/Client.java
+++ b/src/java/org/apache/hadoop/ipc/Client.java
@@ -364,6 +364,35 @@ private synchronized void disposeSasl() {
       }
     }
     
+    private synchronized void setupSaslConnection(final InputStream in2, 
+        final OutputStream out2) 
+    throws javax.security.sasl.SaslException,IOException,InterruptedException {
+      try {
+        saslRpcClient = new SaslRpcClient(authMethod, token,
+            serverPrincipal);
+        saslRpcClient.saslConnect(in2, out2);
+      } catch (javax.security.sasl.SaslException je) {
+        if (authMethod == AuthMethod.KERBEROS && 
+            UserGroupInformation.isLoginKeytabBased()) {
+          //try re-login
+          UserGroupInformation.getCurrentUser().reloginFromKeytab();
+          //try setting up the connection again
+          try {
+            saslRpcClient = new SaslRpcClient(authMethod, token,
+                serverPrincipal);
+            saslRpcClient.saslConnect(in2, out2);
+          } catch (javax.security.sasl.SaslException jee) {
+            UserGroupInformation.
+            setLastUnsuccessfulAuthenticationAttemptTime
+            (System.currentTimeMillis());
+            LOG.warn("Couldn't setup connection for " + 
+                UserGroupInformation.getCurrentUser().getUserName() +
+                " to " + serverPrincipal + " even after relogin.");
+            throw jee;
+          }
+        } else throw je;
+      }
+    }
     /** Connect to the server and set up the I/O streams. It then sends
      * a header to the server and starts
      * the connection thread that waits for responses.
@@ -410,10 +439,8 @@ private synchronized void setupIOstreams() throws InterruptedException {
           }
           ticket.doAs(new PrivilegedExceptionAction<Object>() {
             @Override
-            public Object run() throws IOException {
-              saslRpcClient = new SaslRpcClient(authMethod, token,
-                  serverPrincipal);
-              saslRpcClient.saslConnect(in2, out2);
+            public Object run() throws IOException, InterruptedException {
+              setupSaslConnection(in2, out2);
               return null;
             }
           });
diff --git a/src/java/org/apache/hadoop/security/UserGroupInformation.java b/src/java/org/apache/hadoop/security/UserGroupInformation.java
index 06b3d8baeb..1d49853228 100644
--- a/src/java/org/apache/hadoop/security/UserGroupInformation.java
+++ b/src/java/org/apache/hadoop/security/UserGroupInformation.java
@@ -130,6 +130,10 @@ public boolean logout() throws LoginException {
   private static boolean useKerberos;
   /** Server-side groups fetching service */
   private static Groups groups;
+  /** The last authentication time */
+  private static long lastUnsuccessfulAuthenticationAttemptTime;
+  
+  public static final long MIN_TIME_BEFORE_RELOGIN = 10 * 60 * 1000L;
   
   /**Environment variable pointing to the token cache file*/
   public static final String HADOOP_TOKEN_FILE_LOCATION = 
@@ -202,6 +206,8 @@ public static boolean isSecurityEnabled() {
 
   private final Subject subject;
   
+  private static LoginContext login;
+  
   private static final String OS_LOGIN_MODULE_NAME;
   private static final Class<? extends Principal> OS_PRINCIPAL_CLASS;
   private static final boolean windows = 
@@ -278,6 +284,7 @@ private static class HadoopConfiguration
     static {
       USER_KERBEROS_OPTIONS.put("doNotPrompt", "true");
       USER_KERBEROS_OPTIONS.put("useTicketCache", "true");
+      USER_KERBEROS_OPTIONS.put("renewTGT", "true");
       String ticketCache = System.getenv("KRB5CCNAME");
       if (ticketCache != null) {
         USER_KERBEROS_OPTIONS.put("ticketCache", ticketCache);
@@ -293,8 +300,6 @@ private static class HadoopConfiguration
       KEYTAB_KERBEROS_OPTIONS.put("doNotPrompt", "true");
       KEYTAB_KERBEROS_OPTIONS.put("useKeyTab", "true");
       KEYTAB_KERBEROS_OPTIONS.put("storeKey", "true");
-      KEYTAB_KERBEROS_OPTIONS.put("useTicketCache", "true");
-      KEYTAB_KERBEROS_OPTIONS.put("renewTGT", "true");
     }
     private static final AppConfigurationEntry KEYTAB_KERBEROS_LOGIN =
       new AppConfigurationEntry(Krb5LoginModule.class.getName(),
@@ -355,7 +360,6 @@ public static UserGroupInformation getCurrentUser() throws IOException {
   static UserGroupInformation getLoginUser() throws IOException {
     if (loginUser == null) {
       try {
-        LoginContext login;
         if (isSecurityEnabled()) {
           login = new LoginContext(HadoopConfiguration.USER_KERBEROS_CONFIG_NAME);
         } else {
@@ -391,7 +395,7 @@ static void loginUserFromKeytab(String user,
     keytabFile = path;
     keytabPrincipal = user;
     try {
-      LoginContext login = 
+      login = 
         new LoginContext(HadoopConfiguration.KEYTAB_KERBEROS_CONFIG_NAME);
       login.login();
       loginUser = new UserGroupInformation(login.getSubject());
@@ -400,7 +404,57 @@ static void loginUserFromKeytab(String user,
                             path, le);
     }
   }
+  
+  /**
+   * Re-Login a user in from a keytab file. Loads a user identity from a keytab
+   * file and login them in. They become the currently logged-in user. This
+   * method assumes that {@link #loginUserFromKeytab(String, String)} had 
+   * happened already.
+   * The Subject field of this UserGroupInformation object is updated to have
+   * the new credentials.
+   * @throws IOException on a failure
+   */
+  public synchronized void reloginFromKeytab()
+  throws IOException {
+    if (!isSecurityEnabled())
+      return;
+    if (login == null || keytabFile == null) {
+      throw new IOException("loginUserFromKeyTab must be done first");
+    }
+    if (System.currentTimeMillis() -lastUnsuccessfulAuthenticationAttemptTime <
+          MIN_TIME_BEFORE_RELOGIN) {
+      LOG.warn("Not attempting to re-login since the last re-login was " +
+          "attempted less than " + (MIN_TIME_BEFORE_RELOGIN/1000) + " seconds"+
+          " before.");
+      return;
+    }
+    try {
+      LOG.info("Initiating logout for " + getUserName());
+      //clear up the kerberos state. But the tokens are not cleared! As per 
+      //the Java kerberos login module code, only the kerberos credentials
+      //are cleared
+      login.logout();
+      //login and also update the subject field of this instance to 
+      //have the new credentials (pass it to the LoginContext constructor)
+      login = 
+        new LoginContext(HadoopConfiguration.KEYTAB_KERBEROS_CONFIG_NAME, 
+            getSubject());
+      LOG.info("Initiating re-login for " + keytabPrincipal);
+      login.login();
+    } catch (LoginException le) {
+      throw new IOException("Login failure for " + keytabPrincipal + 
+          " from keytab " + keytabFile, le);
+    } 
+  }
 
+  public synchronized static void 
+    setLastUnsuccessfulAuthenticationAttemptTime(long time) {
+    lastUnsuccessfulAuthenticationAttemptTime = time;
+  }
+  
+  public synchronized static boolean isLoginKeytabBased() {
+    return keytabFile != null;
+  }
   /**
    * Create a user from a login name. It is intended to be used for remote
    * users in RPC, since it won't have any credentials.