fail gracefully on corrupt leveldb files

Author: squito

network/shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

@@ -100,7 +100,19 @@ public ExternalShuffleBlockResolver(TransportConf conf, File registeredExecutorF
             throw new IOException("Unable to create state store", dbExc);
           }
         } else {
-          throw e;
+          logger.error("error opening leveldb file {}.  Creating new file, will not be able to " +
+            "recover state for existing applications", registeredExecutorFile, e);
+          for (File f: registeredExecutorFile.listFiles()) {
+            f.delete();
+          }
+          registeredExecutorFile.delete();
+          options.createIfMissing(true);
+          try {
+            tmpDb = JniDBFactory.factory.open(registeredExecutorFile, options);
+          } catch (NativeDB.DBException dbExc) {
+            throw new IOException("Unable to create state store", dbExc);
+          }
+
         }
       }
       try {

network/yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java

@@ -127,7 +127,7 @@ protected void serviceInit(Configuration conf) {
     try {
       blockHandler = new ExternalShuffleBlockHandler(transportConf, registeredExecutorFile);
     } catch (Exception e) {
-      logger.error("Failed to initial external shuffle service", e);
+      logger.error("Failed to initialize external shuffle service", e);
     }
 
     List<TransportServerBootstrap> bootstraps = Lists.newArrayList();

yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceSuite.scala

@@ -172,7 +172,7 @@ class YarnShuffleServiceSuite extends SparkFunSuite with Matchers with BeforeAnd
     ShuffleTestAccessor.reloadRegisteredExecutors(db) shouldBe empty
   }
 
-  ignore("shuffle service should be robust to corrupt registered executor file") {
+  test("shuffle service should be robust to corrupt registered executor file") {
     s1 = new YarnShuffleService
     s1.init(yarnConfig)
     val app1Id = ApplicationId.newInstance(0, 1)