You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by ni...@apache.org on 2024/04/23 04:10:32 UTC
(celeborn) 03/06: [CELEBORN-1386] LevelDBProvider/RocksDBProvider should create non-existent multi-level directory for LevelDB/RocksDB initialization
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/celeborn.git
commit 657675b36736a4d45189cab2a4f9c5d925f2f686
Author: SteNicholas <pr...@163.com>
AuthorDate: Tue Apr 16 11:37:35 2024 +0800
[CELEBORN-1386] LevelDBProvider/RocksDBProvider should create non-existent multi-level directory for LevelDB/RocksDB initialization
### What changes were proposed in this pull request?
`LevelDBProvider`/`RocksDBProvider` creates non-existent multi-level directory for LevelDB/RocksDB initialization.
### Why are the changes needed?
`RocksDBProvider` creates database if missing via `Options#setCreateIfMissing` when initializing RocksDB at present, which causes the following exception when `dbFile` is non-existent multi-level directory.
```
2024-04-09T03:19:35.6807077Z 24/04/09 03:19:35,679 ERROR [pool-1-thread-1-ScalaTest-running-StorageManagerSuite] RocksDBProvider: error opening rocksdb file /tmp/recover/recovery.rdb. Creating new file, will not be able to recover state for existing applications
2024-04-09T03:19:35.6810066Z org.rocksdb.RocksDBException: While mkdir if missing: /tmp/recover/recovery.rdb: No such file or directory
2024-04-09T03:19:35.6811303Z at org.rocksdb.RocksDB.open(Native Method)
2024-04-09T03:19:35.6812052Z at org.rocksdb.RocksDB.open(RocksDB.java:259)
2024-04-09T03:19:35.6813431Z at org.apache.celeborn.service.deploy.worker.shuffledb.RocksDBProvider.initRockDB(RocksDBProvider.java:66)
2024-04-09T03:19:35.6815230Z at org.apache.celeborn.service.deploy.worker.shuffledb.DBProvider.initDB(DBProvider.java:39)
2024-04-09T03:19:35.6816975Z at org.apache.celeborn.service.deploy.worker.storage.StorageManager.<init>(StorageManager.scala:216)
2024-04-09T03:19:35.6818904Z at org.apache.celeborn.service.deploy.worker.storage.StorageManagerSuite.$anonfun$new$1(StorageManagerSuite.scala:30)
2024-04-09T03:19:35.6820538Z at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
2024-04-09T03:19:35.6821620Z at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
2024-04-09T03:19:35.6822585Z at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
2024-04-09T03:19:35.6823948Z at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
2024-04-09T03:19:35.6824908Z at org.scalatest.Transformer.apply(Transformer.scala:22)
2024-04-09T03:19:35.6825862Z at org.scalatest.Transformer.apply(Transformer.scala:20)
2024-04-09T03:19:35.6827073Z at org.scalatest.funsuite.AnyFunSuiteLike$$anon$1.apply(AnyFunSuiteLike.scala:226)
2024-04-09T03:19:35.6828439Z at org.apache.celeborn.CelebornFunSuite.withFixture(CelebornFunSuite.scala:157)
2024-04-09T03:19:35.6829909Z at org.scalatest.funsuite.AnyFunSuiteLike.invokeWithFixture$1(AnyFunSuiteLike.scala:224)
2024-04-09T03:19:35.6831386Z at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTest$1(AnyFunSuiteLike.scala:236)
2024-04-09T03:19:35.6832590Z at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
2024-04-09T03:19:35.6833727Z at org.scalatest.funsuite.AnyFunSuiteLike.runTest(AnyFunSuiteLike.scala:236)
2024-04-09T03:19:35.6835034Z at org.scalatest.funsuite.AnyFunSuiteLike.runTest$(AnyFunSuiteLike.scala:218)
2024-04-09T03:19:35.6836660Z at org.apache.celeborn.CelebornFunSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6838253Z at org.scalatest.BeforeAndAfterEach.runTest(BeforeAndAfterEach.scala:234)
2024-04-09T03:19:35.6839512Z at org.scalatest.BeforeAndAfterEach.runTest$(BeforeAndAfterEach.scala:227)
2024-04-09T03:19:35.6840766Z at org.apache.celeborn.CelebornFunSuite.runTest(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6842131Z at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTests$1(AnyFunSuiteLike.scala:269)
2024-04-09T03:19:35.6843459Z at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413)
2024-04-09T03:19:35.6844543Z at scala.collection.immutable.List.foreach(List.scala:431)
2024-04-09T03:19:35.6845566Z at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
2024-04-09T03:19:35.6846677Z at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:396)
2024-04-09T03:19:35.6847722Z at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475)
2024-04-09T03:19:35.6849045Z at org.scalatest.funsuite.AnyFunSuiteLike.runTests(AnyFunSuiteLike.scala:269)
2024-04-09T03:19:35.6850358Z at org.scalatest.funsuite.AnyFunSuiteLike.runTests$(AnyFunSuiteLike.scala:268)
2024-04-09T03:19:35.6851608Z at org.scalatest.funsuite.AnyFunSuite.runTests(AnyFunSuite.scala:1564)
2024-04-09T03:19:35.6852566Z at org.scalatest.Suite.run(Suite.scala:1114)
2024-04-09T03:19:35.6853295Z at org.scalatest.Suite.run$(Suite.scala:1096)
2024-04-09T03:19:35.6854857Z at org.scalatest.funsuite.AnyFunSuite.org$scalatest$funsuite$AnyFunSuiteLike$$super$run(AnyFunSuite.scala:1564)
2024-04-09T03:19:35.6856472Z at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$run$1(AnyFunSuiteLike.scala:273)
2024-04-09T03:19:35.6857654Z at org.scalatest.SuperEngine.runImpl(Engine.scala:535)
2024-04-09T03:19:35.6858737Z at org.scalatest.funsuite.AnyFunSuiteLike.run(AnyFunSuiteLike.scala:273)
2024-04-09T03:19:35.6859974Z at org.scalatest.funsuite.AnyFunSuiteLike.run$(AnyFunSuiteLike.scala:272)
2024-04-09T03:19:35.6861519Z at org.apache.celeborn.CelebornFunSuite.org$scalatest$BeforeAndAfterAll$$super$run(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6863041Z at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:213)
2024-04-09T03:19:35.6864233Z at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210)
2024-04-09T03:19:35.6865355Z at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208)
2024-04-09T03:19:35.6866487Z at org.apache.celeborn.CelebornFunSuite.run(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6867764Z at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
2024-04-09T03:19:35.6869119Z at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
2024-04-09T03:19:35.6870146Z at sbt.ForkMain$Run.lambda$runTest$1(ForkMain.java:414)
2024-04-09T03:19:35.6871069Z at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2024-04-09T03:19:35.6872490Z at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
2024-04-09T03:19:35.6873824Z at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
2024-04-09T03:19:35.6874805Z at java.lang.Thread.run(Thread.java:750)
2024-04-09T03:19:35.6887377Z 24/04/09 03:19:35,687 ERROR [pool-1-thread-1-ScalaTest-running-StorageManagerSuite] StorageManager: Init level DB failed:
2024-04-09T03:19:35.6889076Z java.io.IOException: Unable to create state store
2024-04-09T03:19:35.6890473Z at org.apache.celeborn.service.deploy.worker.shuffledb.RocksDBProvider.initRockDB(RocksDBProvider.java:98)
2024-04-09T03:19:35.6894605Z at org.apache.celeborn.service.deploy.worker.shuffledb.DBProvider.initDB(DBProvider.java:39)
2024-04-09T03:19:35.6904452Z at org.apache.celeborn.service.deploy.worker.storage.StorageManager.<init>(StorageManager.scala:216)
2024-04-09T03:19:35.6936013Z at org.apache.celeborn.service.deploy.worker.storage.StorageManagerSuite.$anonfun$new$1(StorageManagerSuite.scala:30)
2024-04-09T03:19:35.6937634Z at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
2024-04-09T03:19:35.6938639Z at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
2024-04-09T03:19:35.6939493Z at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
2024-04-09T03:19:35.6940348Z at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
2024-04-09T03:19:35.6941200Z at org.scalatest.Transformer.apply(Transformer.scala:22)
2024-04-09T03:19:35.6942029Z at org.scalatest.Transformer.apply(Transformer.scala:20)
2024-04-09T03:19:35.6943079Z at org.scalatest.funsuite.AnyFunSuiteLike$$anon$1.apply(AnyFunSuiteLike.scala:226)
2024-04-09T03:19:35.6944350Z at org.apache.celeborn.CelebornFunSuite.withFixture(CelebornFunSuite.scala:157)
2024-04-09T03:19:35.6945683Z at org.scalatest.funsuite.AnyFunSuiteLike.invokeWithFixture$1(AnyFunSuiteLike.scala:224)
2024-04-09T03:19:35.6947057Z at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTest$1(AnyFunSuiteLike.scala:236)
2024-04-09T03:19:35.6948181Z at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
2024-04-09T03:19:35.6949222Z at org.scalatest.funsuite.AnyFunSuiteLike.runTest(AnyFunSuiteLike.scala:236)
2024-04-09T03:19:35.6950415Z at org.scalatest.funsuite.AnyFunSuiteLike.runTest$(AnyFunSuiteLike.scala:218)
2024-04-09T03:19:35.6951915Z at org.apache.celeborn.CelebornFunSuite.org$scalatest$BeforeAndAfterEach$$super$runTest(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6953391Z at org.scalatest.BeforeAndAfterEach.runTest(BeforeAndAfterEach.scala:234)
2024-04-09T03:19:35.6954811Z at org.scalatest.BeforeAndAfterEach.runTest$(BeforeAndAfterEach.scala:227)
2024-04-09T03:19:35.6955990Z at org.apache.celeborn.CelebornFunSuite.runTest(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6957249Z at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTests$1(AnyFunSuiteLike.scala:269)
2024-04-09T03:19:35.6958473Z at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413)
2024-04-09T03:19:35.6959465Z at scala.collection.immutable.List.foreach(List.scala:431)
2024-04-09T03:19:35.6960422Z at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
2024-04-09T03:19:35.6961411Z at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:396)
2024-04-09T03:19:35.6962347Z at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475)
2024-04-09T03:19:35.6963404Z at org.scalatest.funsuite.AnyFunSuiteLike.runTests(AnyFunSuiteLike.scala:269)
2024-04-09T03:19:35.6964635Z at org.scalatest.funsuite.AnyFunSuiteLike.runTests$(AnyFunSuiteLike.scala:268)
2024-04-09T03:19:35.6965797Z at org.scalatest.funsuite.AnyFunSuite.runTests(AnyFunSuite.scala:1564)
2024-04-09T03:19:35.6966679Z at org.scalatest.Suite.run(Suite.scala:1114)
2024-04-09T03:19:35.6967341Z at org.scalatest.Suite.run$(Suite.scala:1096)
2024-04-09T03:19:35.6968835Z at org.scalatest.funsuite.AnyFunSuite.org$scalatest$funsuite$AnyFunSuiteLike$$super$run(AnyFunSuite.scala:1564)
2024-04-09T03:19:35.6970329Z at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$run$1(AnyFunSuiteLike.scala:273)
2024-04-09T03:19:35.6971604Z at org.scalatest.SuperEngine.runImpl(Engine.scala:535)
2024-04-09T03:19:35.6972573Z at org.scalatest.funsuite.AnyFunSuiteLike.run(AnyFunSuiteLike.scala:273)
2024-04-09T03:19:35.6973695Z at org.scalatest.funsuite.AnyFunSuiteLike.run$(AnyFunSuiteLike.scala:272)
2024-04-09T03:19:35.6975279Z at org.apache.celeborn.CelebornFunSuite.org$scalatest$BeforeAndAfterAll$$super$run(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6976986Z at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:213)
2024-04-09T03:19:35.6978212Z at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210)
2024-04-09T03:19:35.6979969Z at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208)
2024-04-09T03:19:35.6982049Z at org.apache.celeborn.CelebornFunSuite.run(CelebornFunSuite.scala:35)
2024-04-09T03:19:35.6982851Z at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
2024-04-09T03:19:35.6983608Z at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
2024-04-09T03:19:35.6984185Z at sbt.ForkMain$Run.lambda$runTest$1(ForkMain.java:414)
2024-04-09T03:19:35.6984688Z at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2024-04-09T03:19:35.6985336Z at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
2024-04-09T03:19:35.6986127Z at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
2024-04-09T03:19:35.6986690Z at java.lang.Thread.run(Thread.java:750)
2024-04-09T03:19:35.6987383Z Caused by: org.rocksdb.RocksDBException: While mkdir if missing: /tmp/recover/recovery.rdb: No such file or directory
2024-04-09T03:19:35.6988110Z at org.rocksdb.RocksDB.open(Native Method)
2024-04-09T03:19:35.6988520Z at org.rocksdb.RocksDB.open(RocksDB.java:259)
2024-04-09T03:19:35.6989257Z at org.apache.celeborn.service.deploy.worker.shuffledb.RocksDBProvider.initRockDB(RocksDBProvider.java:96)
2024-04-09T03:19:35.6989929Z ... 48 more
```
Because `mkdir` does not support creating non-existent multi-level directory, `CreateDirIfMissing` does not support creation of non-existent multi-level directory in [CreateDirIfMissing](https://github.com/facebook/rocksdb/blob/main/env/fs_posix.cc#L637). Meanwhile, `CreateDir` also does not support the creation in [CreateDir](https://github.com/google/leveldb/blob/main/util/env_posix.cc#L625). Therefore `LevelDBProvider`/`RocksDBProvider` should create non-existent multi-level direct [...]
```
IOStatus CreateDirIfMissing(const std::string& name,
const IOOptions& /*opts*/,
IODebugContext* /*dbg*/) override {
if (mkdir(name.c_str(), 0755) != 0) {
if (errno != EEXIST) {
return IOError("While mkdir if missing", name, errno);
} else if (!DirExists(name)) { // Check that name is actually a
// directory.
// Message is taken from mkdir
return IOStatus::IOError("`" + name +
"' exists but is not a directory");
}
}
return IOStatus::OK();
}
```
```
Status CreateDir(const std::string& dirname) override {
if (::mkdir(dirname.c_str(), 0755) != 0) {
return PosixError(dirname, errno);
}
return Status::OK();
}
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`DBProviderSuiteJ#testRockDBCheckVersionFailed`
Closes #2458 from SteNicholas/CELEBORN-1386.
Authored-by: SteNicholas <pr...@163.com>
Signed-off-by: mingji <fe...@alibaba-inc.com>
---
.../deploy/worker/shuffledb/LevelDBProvider.java | 41 ++++++++++++-------
.../deploy/worker/shuffledb/RocksDBProvider.java | 46 ++++++++++++++++------
.../deploy/worker/shuffledb/DBProviderSuiteJ.java | 24 +++++++++--
3 files changed, 81 insertions(+), 30 deletions(-)
diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/LevelDBProvider.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/LevelDBProvider.java
index 7bac20415..839b47dba 100644
--- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/LevelDBProvider.java
+++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/LevelDBProvider.java
@@ -21,6 +21,7 @@ import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import org.apache.commons.io.FileUtils;
import org.fusesource.leveldbjni.JniDBFactory;
import org.fusesource.leveldbjni.internal.NativeDB;
import org.iq80.leveldb.DB;
@@ -38,8 +39,7 @@ import org.apache.celeborn.common.util.PbSerDeUtils;
public class LevelDBProvider {
private static final Logger logger = LoggerFactory.getLogger(LevelDBProvider.class);
- public static org.iq80.leveldb.DB initLevelDB(File dbFile, StoreVersion version)
- throws IOException {
+ public static DB initLevelDB(File dbFile, StoreVersion version) throws IOException {
org.iq80.leveldb.DB tmpDb = null;
if (dbFile != null) {
Options options = new Options();
@@ -50,7 +50,7 @@ public class LevelDBProvider {
} catch (NativeDB.DBException e) {
if (e.isNotFound() || e.getMessage().contains(" does not exist ")) {
logger.info("Creating state database at " + dbFile);
- options.createIfMissing(true);
+ createIfMissing(options, dbFile);
try {
tmpDb = JniDBFactory.factory.open(dbFile, options);
} catch (NativeDB.DBException dbExc) {
@@ -60,21 +60,21 @@ public class LevelDBProvider {
// the leveldb file seems to be corrupt somehow. Lets just blow it away and create a new
// one, so we can keep processing new apps
logger.error(
- "error opening leveldb file {}. Creating new file, will not be able to "
+ "Error opening leveldb file {}. Creating new file, will not be able to "
+ "recover state for existing applications",
dbFile,
e);
if (dbFile.isDirectory()) {
for (File f : dbFile.listFiles()) {
if (!f.delete()) {
- logger.warn("error deleting {}", f.getPath());
+ logger.warn("Error deleting {}", f.getPath());
}
}
}
if (!dbFile.delete()) {
- logger.warn("error deleting {}", dbFile.getPath());
+ logger.warn("Error deleting {}", dbFile.getPath());
}
- options.createIfMissing(true);
+ createIfMissing(options, dbFile);
try {
tmpDb = JniDBFactory.factory.open(dbFile, options);
} catch (NativeDB.DBException dbExc) {
@@ -93,6 +93,19 @@ public class LevelDBProvider {
return tmpDb;
}
+ private static void createIfMissing(Options dbOptions, File dbFile) {
+ logger.info("Creating database file {} if missing", dbFile);
+ dbOptions.createIfMissing(true);
+ // LevelDB does not support creating non-existent multi-level directory.
+ if (!dbFile.exists()) {
+ try {
+ FileUtils.forceMkdir(dbFile);
+ } catch (IOException e) {
+ logger.warn("Failed to create database file {}", dbFile, e);
+ }
+ }
+ }
+
private static class LevelDBLogger implements org.iq80.leveldb.Logger {
private static final Logger LOG = LoggerFactory.getLogger(LevelDBLogger.class);
@@ -107,26 +120,26 @@ public class LevelDBProvider {
* Minor version differences are allowed -- meaning we should be able to read dbs that are either
* earlier *or* later on the minor version.
*/
- public static void checkVersion(DB db, StoreVersion newversion) throws IOException {
+ public static void checkVersion(DB db, StoreVersion newVersion) throws IOException {
byte[] bytes = db.get(StoreVersion.KEY);
if (bytes == null) {
- storeVersion(db, newversion);
+ storeVersion(db, newVersion);
} else {
ArrayList<Integer> versions = PbSerDeUtils.fromPbStoreVersion(bytes);
StoreVersion version = new StoreVersion(versions.get(0), versions.get(1));
- if (version.major != newversion.major) {
+ if (version.major != newVersion.major) {
throw new IOException(
- "cannot read state DB with version "
+ "Cannot read state DB with version "
+ version
+ ", incompatible "
+ "with current version "
- + newversion);
+ + newVersion);
}
- storeVersion(db, newversion);
+ storeVersion(db, newVersion);
}
}
- public static void storeVersion(DB db, StoreVersion version) throws IOException {
+ public static void storeVersion(DB db, StoreVersion version) {
db.put(StoreVersion.KEY, PbSerDeUtils.toPbStoreVersion(version.major, version.minor));
}
}
diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBProvider.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBProvider.java
index 4229008c0..75f728a30 100644
--- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBProvider.java
+++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/shuffledb/RocksDBProvider.java
@@ -22,7 +22,14 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Objects;
-import org.rocksdb.*;
+import org.apache.commons.io.FileUtils;
+import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
+import org.rocksdb.CompressionType;
+import org.rocksdb.InfoLogLevel;
+import org.rocksdb.Options;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Status;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,7 +74,7 @@ public class RocksDBProvider {
} catch (RocksDBException e) {
if (e.getStatus().getCode() == Status.Code.NotFound) {
logger.info("Creating state database at " + dbFile);
- dbOptions.setCreateIfMissing(true);
+ createIfMissing(dbOptions, dbFile);
try {
tmpDb = org.rocksdb.RocksDB.open(dbOptions, dbFile.toString());
} catch (RocksDBException dbExc) {
@@ -77,21 +84,21 @@ public class RocksDBProvider {
// the RocksDB file seems to be corrupt somehow. Let's just blow it away and create
// a new one, so we can keep processing new apps
logger.error(
- "error opening rocksdb file {}. Creating new file, will not be able to "
+ "Error opening rocksdb file {}. Creating new file, will not be able to "
+ "recover state for existing applications",
dbFile,
e);
if (dbFile.isDirectory()) {
for (File f : Objects.requireNonNull(dbFile.listFiles())) {
if (!f.delete()) {
- logger.warn("error deleting {}", f.getPath());
+ logger.warn("Error deleting {}", f.getPath());
}
}
}
if (!dbFile.delete()) {
- logger.warn("error deleting {}", dbFile.getPath());
+ logger.warn("Error deleting {}", dbFile.getPath());
}
- dbOptions.setCreateIfMissing(true);
+ createIfMissing(dbOptions, dbFile);
try {
tmpDb = org.rocksdb.RocksDB.open(dbOptions, dbFile.toString());
} catch (RocksDBException dbExc) {
@@ -114,6 +121,19 @@ public class RocksDBProvider {
return tmpDb;
}
+ private static void createIfMissing(Options dbOptions, File dbFile) {
+ logger.info("Creating database file {} if missing", dbFile);
+ dbOptions.setCreateIfMissing(true);
+ // RocksDB does not support creating non-existent multi-level directory.
+ if (!dbFile.exists()) {
+ try {
+ FileUtils.forceMkdir(dbFile);
+ } catch (IOException e) {
+ logger.warn("Failed to create database file {}", dbFile, e);
+ }
+ }
+ }
+
private static class RocksDBLogger extends org.rocksdb.Logger {
private static final Logger LOG = LoggerFactory.getLogger(RocksDBLogger.class);
@@ -134,28 +154,28 @@ public class RocksDBProvider {
* Minor version differences are allowed -- meaning we should be able to read dbs that are either
* earlier *or* later on the minor version.
*/
- public static void checkVersion(org.rocksdb.RocksDB db, StoreVersion newversion)
+ public static void checkVersion(org.rocksdb.RocksDB db, StoreVersion newVersion)
throws IOException, RocksDBException {
byte[] bytes = db.get(StoreVersion.KEY);
if (bytes == null) {
- storeVersion(db, newversion);
+ storeVersion(db, newVersion);
} else {
ArrayList<Integer> versions = PbSerDeUtils.fromPbStoreVersion(bytes);
StoreVersion version = new StoreVersion(versions.get(0), versions.get(1));
- if (version.major != newversion.major) {
+ if (version.major != newVersion.major) {
throw new IOException(
- "cannot read state DB with version "
+ "Cannot read state DB with version "
+ version
+ ", incompatible "
+ "with current version "
- + newversion);
+ + newVersion);
}
- storeVersion(db, newversion);
+ storeVersion(db, newVersion);
}
}
public static void storeVersion(org.rocksdb.RocksDB db, StoreVersion version)
- throws IOException, RocksDBException {
+ throws RocksDBException {
db.put(StoreVersion.KEY, PbSerDeUtils.toPbStoreVersion(version.major, version.minor));
}
}
diff --git a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/shuffledb/DBProviderSuiteJ.java b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/shuffledb/DBProviderSuiteJ.java
index e9399210b..69f0aa190 100644
--- a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/shuffledb/DBProviderSuiteJ.java
+++ b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/shuffledb/DBProviderSuiteJ.java
@@ -23,15 +23,30 @@ import static org.junit.Assume.assumeFalse;
import java.io.File;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.UUID;
import org.apache.commons.lang3.SystemUtils;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
import org.apache.celeborn.common.util.JavaUtils;
import org.apache.celeborn.common.util.Utils;
+@RunWith(Parameterized.class)
public class DBProviderSuiteJ {
+ @Parameter public boolean createDirectory;
+
+ @Parameters(name = "createDirectory: {0}")
+ public static Collection<Object> data() {
+ return Arrays.asList(new Object[] {true, false});
+ }
+
@Test
public void testRockDBCheckVersionFailed() throws IOException {
testCheckVersionFailed(DBBackend.ROCKSDB, "rocksdb");
@@ -44,8 +59,11 @@ public class DBProviderSuiteJ {
}
private void testCheckVersionFailed(DBBackend dbBackend, String namePrefix) throws IOException {
- String root = System.getProperty("java.io.tmpdir");
- File dbFile = Utils.createDirectory(root, namePrefix);
+ File dbDir = new File(System.getProperty("java.io.tmpdir"), namePrefix);
+ File dbFile =
+ createDirectory
+ ? Utils.createDirectory(dbDir.getPath(), namePrefix)
+ : new File(dbDir.getPath(), String.format("%s-%s", namePrefix, UUID.randomUUID()));
try {
StoreVersion v1 = new StoreVersion(1, 0);
DBProvider.initDB(dbBackend, dbFile, v1).close();
@@ -54,7 +72,7 @@ public class DBProviderSuiteJ {
assertThrows(IOException.class, () -> DBProvider.initDB(dbBackend, dbFile, v2));
assertTrue(ioe.getMessage().contains("incompatible with current version StoreVersion[2.0]"));
} finally {
- JavaUtils.deleteRecursively(dbFile);
+ JavaUtils.deleteRecursively(dbDir);
}
}
}