YARN-2958. Made RMStateStore not update the last sequence number when…

… updating the delegation token. Contributed by Varun Saxena.

hadoop-yarn-project/CHANGES.txt

@@ -311,6 +311,9 @@ Release 2.7.0 - UNRELEASED
     YARN-2922. ConcurrentModificationException in CapacityScheduler's LeafQueue.
     (Rohith Sharmaks via ozawa)
 
+    YARN-2958. Made RMStateStore not update the last sequence number when updating the
+    delegation token. (Varun Saxena via zjshen)
+
 Release 2.6.0 - 2014-11-18
 
   INCOMPATIBLE CHANGES

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java

@@ -60,8 +60,6 @@
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationAttemptStateDataPBImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.ApplicationStateDataPBImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.impl.pb.EpochPBImpl;
-import org.apache.hadoop.yarn.util.ConverterUtils;
-
 import com.google.common.annotations.VisibleForTesting;
 
 @Private
@@ -452,11 +450,10 @@ public synchronized void removeApplicationStateInternal(
   }
 
   @Override
-  public synchronized void storeRMDelegationTokenAndSequenceNumberState(
-      RMDelegationTokenIdentifier identifier, Long renewDate,
-      int latestSequenceNumber) throws Exception {
-    storeOrUpdateRMDelegationTokenAndSequenceNumberState(
-        identifier, renewDate,latestSequenceNumber, false);
+  public synchronized void storeRMDelegationTokenState(
+      RMDelegationTokenIdentifier identifier, Long renewDate)
+      throws Exception {
+    storeOrUpdateRMDelegationTokenState(identifier, renewDate, false);
   }
 
   @Override
@@ -469,16 +466,15 @@ public synchronized void removeRMDelegationTokenState(
   }
 
   @Override
-  protected void updateRMDelegationTokenAndSequenceNumberInternal(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception {
-    storeOrUpdateRMDelegationTokenAndSequenceNumberState(
-        rmDTIdentifier, renewDate,latestSequenceNumber, true);
+  protected void updateRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception {
+    storeOrUpdateRMDelegationTokenState(rmDTIdentifier, renewDate, true);
   }
 
-  private void storeOrUpdateRMDelegationTokenAndSequenceNumberState(
+  private void storeOrUpdateRMDelegationTokenState(
       RMDelegationTokenIdentifier identifier, Long renewDate,
-      int latestSequenceNumber, boolean isUpdate) throws Exception {
+      boolean isUpdate) throws Exception {
     Path nodeCreatePath =
         getNodePath(rmDTSecretManagerRoot,
           DELEGATION_TOKEN_PREFIX + identifier.getSequenceNumber());
@@ -490,23 +486,24 @@ private void storeOrUpdateRMDelegationTokenAndSequenceNumberState(
     } else {
       LOG.info("Storing RMDelegationToken_" + identifier.getSequenceNumber());
       writeFile(nodeCreatePath, identifierData.toByteArray());
-    }
 
-    // store sequence number
-    Path latestSequenceNumberPath = getNodePath(rmDTSecretManagerRoot,
-          DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX + latestSequenceNumber);
-    LOG.info("Storing " + DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX
-        + latestSequenceNumber);
-    if (dtSequenceNumberPath == null) {
-      if (!createFile(latestSequenceNumberPath)) {
-        throw new Exception("Failed to create " + latestSequenceNumberPath);
-      }
-    } else {
-      if (!renameFile(dtSequenceNumberPath, latestSequenceNumberPath)) {
-        throw new Exception("Failed to rename " + dtSequenceNumberPath);
+      // store sequence number
+      Path latestSequenceNumberPath = getNodePath(rmDTSecretManagerRoot,
+            DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX
+            + identifier.getSequenceNumber());
+      LOG.info("Storing " + DELEGATION_TOKEN_SEQUENCE_NUMBER_PREFIX
+          + identifier.getSequenceNumber());
+      if (dtSequenceNumberPath == null) {
+        if (!createFile(latestSequenceNumberPath)) {
+          throw new Exception("Failed to create " + latestSequenceNumberPath);
+        }
+      } else {
+        if (!renameFile(dtSequenceNumberPath, latestSequenceNumberPath)) {
+          throw new Exception("Failed to rename " + dtSequenceNumberPath);
+        }
       }
+      dtSequenceNumberPath = latestSequenceNumberPath;
     }
-    dtSequenceNumberPath = latestSequenceNumberPath;
   }
 
   @Override

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/LeveldbRMStateStore.java

@@ -544,31 +544,30 @@ protected void removeApplicationStateInternal(ApplicationStateData appState)
       throw new IOException(e);
     }
   }
-
-  @Override
-  protected void storeRMDelegationTokenAndSequenceNumberState(
-      RMDelegationTokenIdentifier tokenId, Long renewDate,
-      int latestSequenceNumber) throws IOException {
+
+  private void storeOrUpdateRMDT(RMDelegationTokenIdentifier tokenId,
+      Long renewDate, boolean isUpdate) throws IOException {
     String tokenKey = getRMDTTokenNodeKey(tokenId);
     RMDelegationTokenIdentifierData tokenData =
         new RMDelegationTokenIdentifierData(tokenId, renewDate);
-    ByteArrayOutputStream bs = new ByteArrayOutputStream();
-    DataOutputStream ds = new DataOutputStream(bs);
-    try {
-      ds.writeInt(latestSequenceNumber);
-    } finally {
-      ds.close();
-    }
     if (LOG.isDebugEnabled()) {
       LOG.debug("Storing token to " + tokenKey);
-      LOG.debug("Storing " + latestSequenceNumber + " to "
-          + RM_DT_SEQUENCE_NUMBER_KEY);
     }
     try {
       WriteBatch batch = db.createWriteBatch();
       try {
         batch.put(bytes(tokenKey), tokenData.toByteArray());
-        batch.put(bytes(RM_DT_SEQUENCE_NUMBER_KEY), bs.toByteArray());
+        if(!isUpdate) {
+          ByteArrayOutputStream bs = new ByteArrayOutputStream();
+          try (DataOutputStream ds = new DataOutputStream(bs)) {
+            ds.writeInt(tokenId.getSequenceNumber());
+          }
+          if (LOG.isDebugEnabled()) {
+            LOG.debug("Storing " + tokenId.getSequenceNumber() + " to "
+                + RM_DT_SEQUENCE_NUMBER_KEY);   
+          }
+          batch.put(bytes(RM_DT_SEQUENCE_NUMBER_KEY), bs.toByteArray());
+        }
         db.write(batch);
       } finally {
         batch.close();
@@ -579,11 +578,17 @@ protected void storeRMDelegationTokenAndSequenceNumberState(
   }
 
   @Override
-  protected void updateRMDelegationTokenAndSequenceNumberInternal(
-      RMDelegationTokenIdentifier tokenId, Long renewDate,
-      int latestSequenceNumber) throws IOException {
-    storeRMDelegationTokenAndSequenceNumberState(tokenId, renewDate,
-        latestSequenceNumber);
+  protected void storeRMDelegationTokenState(
+      RMDelegationTokenIdentifier tokenId, Long renewDate)
+      throws IOException {
+    storeOrUpdateRMDT(tokenId, renewDate, false);
+  }
+
+  @Override
+  protected void updateRMDelegationTokenState(
+      RMDelegationTokenIdentifier tokenId, Long renewDate)
+      throws IOException {
+    storeOrUpdateRMDT(tokenId, renewDate, true);
   }
 
   @Override

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java

@@ -149,23 +149,30 @@ public synchronized void removeApplicationStateInternal(
     }
   }
 
-  @Override
-  public synchronized void storeRMDelegationTokenAndSequenceNumberState(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception {
+  private void storeOrUpdateRMDT(RMDelegationTokenIdentifier rmDTIdentifier,
+      Long renewDate, boolean isUpdate) throws Exception {
     Map<RMDelegationTokenIdentifier, Long> rmDTState =
         state.rmSecretManagerState.getTokenState();
     if (rmDTState.containsKey(rmDTIdentifier)) {
       IOException e = new IOException("RMDelegationToken: " + rmDTIdentifier
-              + "is already stored.");
+          + "is already stored.");
       LOG.info("Error storing info for RMDelegationToken: " + rmDTIdentifier, e);
       throw e;
     }
     rmDTState.put(rmDTIdentifier, renewDate);
-    state.rmSecretManagerState.dtSequenceNumber = latestSequenceNumber;
+    if(!isUpdate) {
+      state.rmSecretManagerState.dtSequenceNumber = 
+          rmDTIdentifier.getSequenceNumber();
+    }
     LOG.info("Store RMDT with sequence number "
-        + rmDTIdentifier.getSequenceNumber()
-        + ". And the latest sequence number is " + latestSequenceNumber);
+             + rmDTIdentifier.getSequenceNumber());
+  }
+
+  @Override
+  public synchronized void storeRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception {
+    storeOrUpdateRMDT(rmDTIdentifier, renewDate, false);
   }
 
   @Override
@@ -179,12 +186,11 @@ public synchronized void removeRMDelegationTokenState(
   }
 
   @Override
-  protected void updateRMDelegationTokenAndSequenceNumberInternal(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception {
+  protected void updateRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception {
     removeRMDelegationTokenState(rmDTIdentifier);
-    storeRMDelegationTokenAndSequenceNumberState(
-        rmDTIdentifier, renewDate, latestSequenceNumber);
+    storeOrUpdateRMDT(rmDTIdentifier, renewDate, true);
     LOG.info("Update RMDT with sequence number "
         + rmDTIdentifier.getSequenceNumber());
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/NullRMStateStore.java

@@ -77,9 +77,9 @@ protected void removeApplicationStateInternal(ApplicationStateData appState)
   }
 
   @Override
-  public void storeRMDelegationTokenAndSequenceNumberState(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception {
+  public void storeRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception {
     // Do nothing
   }
 
@@ -90,9 +90,9 @@ public void removeRMDelegationTokenState(RMDelegationTokenIdentifier rmDTIdentif
   }
 
   @Override
-  protected void updateRMDelegationTokenAndSequenceNumberInternal(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception {
+  protected void updateRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception {
     // Do nothing
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java

@@ -296,9 +296,8 @@ public void transition(RMStateStore store, RMStateStoreEvent event) {
       RMStateStoreRMDTEvent dtEvent = (RMStateStoreRMDTEvent) event;
       try {
         LOG.info("Storing RMDelegationToken and SequenceNumber");
-        store.storeRMDelegationTokenAndSequenceNumberState(
-            dtEvent.getRmDTIdentifier(), dtEvent.getRenewDate(),
-            dtEvent.getLatestSequenceNumber());
+        store.storeRMDelegationTokenState(
+            dtEvent.getRmDTIdentifier(), dtEvent.getRenewDate());
       } catch (Exception e) {
         LOG.error("Error While Storing RMDelegationToken and SequenceNumber ",
             e);
@@ -341,9 +340,8 @@ public void transition(RMStateStore store, RMStateStoreEvent event) {
       RMStateStoreRMDTEvent dtEvent = (RMStateStoreRMDTEvent) event;
       try {
         LOG.info("Updating RMDelegationToken and SequenceNumber");
-        store.updateRMDelegationTokenAndSequenceNumberInternal(
-            dtEvent.getRmDTIdentifier(), dtEvent.getRenewDate(),
-            dtEvent.getLatestSequenceNumber());
+        store.updateRMDelegationTokenState(
+            dtEvent.getRmDTIdentifier(), dtEvent.getRenewDate());
       } catch (Exception e) {
         LOG.error("Error While Updating RMDelegationToken and SequenceNumber ",
             e);
@@ -672,29 +670,28 @@ protected abstract void updateApplicationAttemptStateInternal(
    * RMDTSecretManager call this to store the state of a delegation token
    * and sequence number
    */
-  public void storeRMDelegationTokenAndSequenceNumber(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) {
+  public void storeRMDelegationToken(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate) {
     handleStoreEvent(new RMStateStoreRMDTEvent(rmDTIdentifier, renewDate,
-        latestSequenceNumber, RMStateStoreEventType.STORE_DELEGATION_TOKEN));
+        RMStateStoreEventType.STORE_DELEGATION_TOKEN));
   }
 
   /**
    * Blocking API
    * Derived classes must implement this method to store the state of
    * RMDelegationToken and sequence number
    */
-  protected abstract void storeRMDelegationTokenAndSequenceNumberState(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception;
+  protected abstract void storeRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception;
 
   /**
    * RMDTSecretManager call this to remove the state of a delegation token
    */
   public void removeRMDelegationToken(
-      RMDelegationTokenIdentifier rmDTIdentifier, int sequenceNumber) {
+      RMDelegationTokenIdentifier rmDTIdentifier) {
     handleStoreEvent(new RMStateStoreRMDTEvent(rmDTIdentifier, null,
-        sequenceNumber, RMStateStoreEventType.REMOVE_DELEGATION_TOKEN));
+        RMStateStoreEventType.REMOVE_DELEGATION_TOKEN));
   }
 
   /**
@@ -708,21 +705,20 @@ protected abstract void removeRMDelegationTokenState(
    * RMDTSecretManager call this to update the state of a delegation token
    * and sequence number
    */
-  public void updateRMDelegationTokenAndSequenceNumber(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) {
+  public void updateRMDelegationToken(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate) {
     handleStoreEvent(new RMStateStoreRMDTEvent(rmDTIdentifier, renewDate,
-        latestSequenceNumber, RMStateStoreEventType.UPDATE_DELEGATION_TOKEN));
+        RMStateStoreEventType.UPDATE_DELEGATION_TOKEN));
   }
 
   /**
    * Blocking API
    * Derived classes must implement this method to update the state of
    * RMDelegationToken and sequence number
    */
-  protected abstract void updateRMDelegationTokenAndSequenceNumberInternal(
-      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate,
-      int latestSequenceNumber) throws Exception;
+  protected abstract void updateRMDelegationTokenState(
+      RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
+      throws Exception;
 
   /**
    * RMDTSecretManager call this to store the state of a master key

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreRMDTEvent.java

@@ -23,18 +23,16 @@
 public class RMStateStoreRMDTEvent extends RMStateStoreEvent {
   private RMDelegationTokenIdentifier rmDTIdentifier;
   private Long renewDate;
-  private int latestSequenceNumber;
 
   public RMStateStoreRMDTEvent(RMStateStoreEventType type) {
     super(type);
   }
 
   public RMStateStoreRMDTEvent(RMDelegationTokenIdentifier rmDTIdentifier,
-      Long renewDate, int latestSequenceNumber, RMStateStoreEventType type) {
+      Long renewDate, RMStateStoreEventType type) {
     this(type);
     this.rmDTIdentifier = rmDTIdentifier;
     this.renewDate = renewDate;
-    this.latestSequenceNumber = latestSequenceNumber;
   }
 
   public RMDelegationTokenIdentifier getRmDTIdentifier() {
@@ -44,8 +42,4 @@ public RMDelegationTokenIdentifier getRmDTIdentifier() {
   public Long getRenewDate() {
     return renewDate;
   }
-
-  public int getLatestSequenceNumber() {
-    return latestSequenceNumber;
-  }
-}
+}