YARN-6354. LeveldbRMStateStore can parse invalid keys when recovering reservations. Contributed by Jason Lowe

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

@@ -214,6 +214,11 @@ public class LeveldbRMStateStore extends RMStateStore {
     return db == null;
   }
 
+  @VisibleForTesting
+  DB getDatabase() {
+    return db;
+  }
+
   @Override
   protected Version loadVersion() throws Exception {
     Version version = null;
@@ -284,6 +289,9 @@ public class LeveldbRMStateStore extends RMStateStore {
       while (iter.hasNext()) {
         Entry<byte[],byte[]> entry = iter.next();
         String key = asString(entry.getKey());
+        if (!key.startsWith(RM_RESERVATION_KEY_PREFIX)) {
+          break;
+        }
 
         String planReservationString =
             key.substring(RM_RESERVATION_KEY_PREFIX.length());

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestLeveldbRMStateStore.java

@@ -31,6 +31,7 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.records.Version;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
+import org.fusesource.leveldbjni.JniDBFactory;
 import org.iq80.leveldb.DB;
 import org.junit.After;
 import org.junit.Before;
@@ -125,17 +126,27 @@ public class TestLeveldbRMStateStore extends RMStateStoreTestBase {
   public void testCompactionCycle() throws Exception {
     final DB mockdb = mock(DB.class);
     conf.setLong(YarnConfiguration.RM_LEVELDB_COMPACTION_INTERVAL_SECS, 1);
-    LeveldbRMStateStore store = new LeveldbRMStateStore() {
+    stateStore = new LeveldbRMStateStore() {
       @Override
       protected DB openDatabase() throws Exception {
         return mockdb;
       }
     };
-    store.init(conf);
-    store.start();
+    stateStore.init(conf);
+    stateStore.start();
     verify(mockdb, timeout(10000)).compactRange(
         (byte[]) isNull(), (byte[]) isNull());
-    store.close();
+  }
+
+  @Test
+  public void testBadKeyIteration() throws Exception {
+    stateStore = new LeveldbRMStateStore();
+    stateStore.init(conf);
+    stateStore.start();
+    DB db = stateStore.getDatabase();
+    // add an entry that appears at the end of the database when iterating
+    db.put(JniDBFactory.bytes("zzz"), JniDBFactory.bytes("z"));
+    stateStore.loadState();
   }
 
   class LeveldbStateStoreTester implements RMStateStoreHelper {