diff --git a/HISTORY.md b/HISTORY.md
index cdd5c77e9..762e57f05 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -14,6 +14,9 @@
 ### New Features
 * Add FileSystem::ReadAsync API in io_tracing
 
+### Behavior changes
+* DB::Open(), DB::OpenAsSecondary() will fail if a Logger cannot be created (#9984)
+
 ## 7.3.0 (05/20/2022)
 ### Bug Fixes
 * Fixed a bug where manual flush would block forever even though flush options had wait=false.
diff --git a/db/db_basic_test.cc b/db/db_basic_test.cc
index a6bcea43b..32b1c8813 100644
--- a/db/db_basic_test.cc
+++ b/db/db_basic_test.cc
@@ -4053,6 +4053,27 @@ TEST_F(DBBasicTest, DestroyDefaultCfHandle) {
   ASSERT_TRUE(db_->DestroyColumnFamilyHandle(default_cf).IsInvalidArgument());
 }
 
+TEST_F(DBBasicTest, FailOpenIfLoggerCreationFail) {
+  Options options = GetDefaultOptions();
+  options.create_if_missing = true;
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->SetCallBack(
+      "rocksdb::CreateLoggerFromOptions:AfterGetPath", [&](void* arg) {
+        auto* s = reinterpret_cast<Status*>(arg);
+        assert(s);
+        *s = Status::IOError("Injected");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  Status s = TryReopen(options);
+  ASSERT_EQ(nullptr, options.info_log);
+  ASSERT_TRUE(s.IsAborted());
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 #ifndef ROCKSDB_LITE
 TEST_F(DBBasicTest, VerifyFileChecksums) {
   Options options = GetDefaultOptions();
diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc
index 9004946f4..294df6531 100644
--- a/db/db_impl/db_impl_open.cc
+++ b/db/db_impl/db_impl_open.cc
@@ -1733,6 +1733,11 @@ Status DBImpl::Open(const DBOptions& db_options, const std::string& dbname,
   }
 
   DBImpl* impl = new DBImpl(db_options, dbname, seq_per_batch, batch_per_txn);
+  if (!impl->immutable_db_options_.info_log) {
+    s = Status::Aborted("Failed to create logger");
+    delete impl;
+    return s;
+  }
   s = impl->env_->CreateDirIfMissing(impl->immutable_db_options_.GetWalDir());
   if (s.ok()) {
     std::vector<std::string> paths;
diff --git a/db/db_impl/db_impl_secondary.cc b/db/db_impl/db_impl_secondary.cc
index 5e9041d69..980638158 100644
--- a/db/db_impl/db_impl_secondary.cc
+++ b/db/db_impl/db_impl_secondary.cc
@@ -689,6 +689,7 @@ Status DB::OpenAsSecondary(
     s = CreateLoggerFromOptions(secondary_path, tmp_opts, &tmp_opts.info_log);
     if (!s.ok()) {
       tmp_opts.info_log = nullptr;
+      return s;
     }
   }
 
diff --git a/db/db_secondary_test.cc b/db/db_secondary_test.cc
index de06bd560..930ff468b 100644
--- a/db/db_secondary_test.cc
+++ b/db/db_secondary_test.cc
@@ -123,6 +123,30 @@ class DBSecondaryTest : public DBSecondaryTestBase {
   explicit DBSecondaryTest() : DBSecondaryTestBase("db_secondary_test") {}
 };
 
+TEST_F(DBSecondaryTest, FailOpenIfLoggerCreationFail) {
+  Options options = GetDefaultOptions();
+  options.create_if_missing = true;
+  Reopen(options);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  SyncPoint::GetInstance()->SetCallBack(
+      "rocksdb::CreateLoggerFromOptions:AfterGetPath", [&](void* arg) {
+        auto* s = reinterpret_cast<Status*>(arg);
+        assert(s);
+        *s = Status::IOError("Injected");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  options.max_open_files = -1;
+  Status s = TryOpenSecondary(options);
+  ASSERT_EQ(nullptr, options.info_log);
+  ASSERT_TRUE(s.IsIOError());
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 TEST_F(DBSecondaryTest, NonExistingDb) {
   Destroy(last_options_);
 
diff --git a/java/src/test/java/org/rocksdb/RocksMemEnvTest.java b/java/src/test/java/org/rocksdb/RocksMemEnvTest.java
index a03a0f0ae..cce0c61e0 100644
--- a/java/src/test/java/org/rocksdb/RocksMemEnvTest.java
+++ b/java/src/test/java/org/rocksdb/RocksMemEnvTest.java
@@ -38,7 +38,7 @@ public class RocksMemEnvTest {
          final FlushOptions flushOptions = new FlushOptions()
              .setWaitForFlush(true);
     ) {
-      try (final RocksDB db = RocksDB.open(options, "dir/db")) {
+      try (final RocksDB db = RocksDB.open(options, "/dir/db")) {
         // write key/value pairs using MemEnv
         for (int i = 0; i < keys.length; i++) {
           db.put(keys[i], values[i]);
@@ -75,7 +75,7 @@ public class RocksMemEnvTest {
 
       // After reopen the values shall still be in the mem env.
       // as long as the env is not freed.
-      try (final RocksDB db = RocksDB.open(options, "dir/db")) {
+      try (final RocksDB db = RocksDB.open(options, "/dir/db")) {
         // read key/value pairs using MemEnv
         for (int i = 0; i < keys.length; i++) {
           assertThat(db.get(keys[i])).isEqualTo(values[i]);
@@ -106,12 +106,9 @@ public class RocksMemEnvTest {
     };
 
     try (final Env env = new RocksMemEnv(Env.getDefault());
-         final Options options = new Options()
-             .setCreateIfMissing(true)
-             .setEnv(env);
-         final RocksDB db = RocksDB.open(options, "dir/db");
-         final RocksDB otherDb = RocksDB.open(options, "dir/otherDb")
-    ) {
+         final Options options = new Options().setCreateIfMissing(true).setEnv(env);
+         final RocksDB db = RocksDB.open(options, "/dir/db");
+         final RocksDB otherDb = RocksDB.open(options, "/dir/otherDb")) {
       // write key/value pairs using MemEnv
       // to db and to otherDb.
       for (int i = 0; i < keys.length; i++) {
@@ -135,10 +132,8 @@ public class RocksMemEnvTest {
   @Test(expected = RocksDBException.class)
   public void createIfMissingFalse() throws RocksDBException {
     try (final Env env = new RocksMemEnv(Env.getDefault());
-         final Options options = new Options()
-             .setCreateIfMissing(false)
-             .setEnv(env);
-         final RocksDB db = RocksDB.open(options, "db/dir")) {
+         final Options options = new Options().setCreateIfMissing(false).setEnv(env);
+         final RocksDB db = RocksDB.open(options, "/db/dir")) {
       // shall throw an exception because db dir does not
       // exist.
     }
diff --git a/logging/auto_roll_logger.cc b/logging/auto_roll_logger.cc
index d32645a42..f7929434e 100644
--- a/logging/auto_roll_logger.cc
+++ b/logging/auto_roll_logger.cc
@@ -270,6 +270,7 @@ Status CreateLoggerFromOptions(const std::string& dbname,
   Env* env = options.env;
   std::string db_absolute_path;
   Status s = env->GetAbsolutePath(dbname, &db_absolute_path);
+  TEST_SYNC_POINT_CALLBACK("rocksdb::CreateLoggerFromOptions:AfterGetPath", &s);
   if (!s.ok()) {
     return s;
   }
@@ -277,10 +278,20 @@ Status CreateLoggerFromOptions(const std::string& dbname,
       InfoLogFileName(dbname, db_absolute_path, options.db_log_dir);
 
   const auto& clock = env->GetSystemClock();
-  env->CreateDirIfMissing(dbname)
-      .PermitUncheckedError();  // In case it does not exist
-  // Currently we only support roll by time-to-roll and log size
+  // In case it does not exist
+  s = env->CreateDirIfMissing(dbname);
+  if (!s.ok()) {
+    return s;
+  }
+
+  if (!options.db_log_dir.empty()) {
+    s = env->CreateDirIfMissing(options.db_log_dir);
+    if (!s.ok()) {
+      return s;
+    }
+  }
 #ifndef ROCKSDB_LITE
+  // Currently we only support roll by time-to-roll and log size
   if (options.log_file_time_to_roll > 0 || options.max_log_file_size > 0) {
     AutoRollLogger* result = new AutoRollLogger(
         env->GetFileSystem(), clock, dbname, options.db_log_dir,
diff --git a/logging/auto_roll_logger_test.cc b/logging/auto_roll_logger_test.cc
index 19e7ea43f..89beaaf50 100644
--- a/logging/auto_roll_logger_test.cc
+++ b/logging/auto_roll_logger_test.cc
@@ -50,18 +50,28 @@ void LogMessage(const InfoLogLevel log_level, Logger* logger,
 class AutoRollLoggerTest : public testing::Test {
  public:
   static void InitTestDb() {
+    // TODO replace the `system` calls with Env/FileSystem APIs.
 #ifdef OS_WIN
     // Replace all slashes in the path so windows CompSpec does not
     // become confused
+    std::string testDbDir(kTestDbDir);
+    std::replace_if(
+        testDbDir.begin(), testDbDir.end(), [](char ch) { return ch == '/'; },
+        '\\');
+    std::string deleteDbDirCmd =
+        "if exist " + testDbDir + " rd /s /q " + testDbDir;
+    ASSERT_TRUE(system(deleteDbDirCmd.c_str()) == 0);
+
     std::string testDir(kTestDir);
     std::replace_if(testDir.begin(), testDir.end(),
                     [](char ch) { return ch == '/'; }, '\\');
     std::string deleteCmd = "if exist " + testDir + " rd /s /q " + testDir;
 #else
-    std::string deleteCmd = "rm -rf " + kTestDir;
+    std::string deleteCmd = "rm -rf " + kTestDir + " " + kTestDbDir;
 #endif
     ASSERT_TRUE(system(deleteCmd.c_str()) == 0);
     ASSERT_OK(Env::Default()->CreateDir(kTestDir));
+    ASSERT_OK(Env::Default()->CreateDir(kTestDbDir));
   }
 
   void RollLogFileBySizeTest(AutoRollLogger* logger, size_t log_max_size,
@@ -108,6 +118,7 @@ class AutoRollLoggerTest : public testing::Test {
 
   static const std::string kSampleMessage;
   static const std::string kTestDir;
+  static const std::string kTestDbDir;
   static const std::string kLogFile;
   static Env* default_env;
 };
@@ -116,6 +127,8 @@ const std::string AutoRollLoggerTest::kSampleMessage(
     "this is the message to be written to the log file!!");
 const std::string AutoRollLoggerTest::kTestDir(
     test::PerThreadDBPath("db_log_test"));
+const std::string AutoRollLoggerTest::kTestDbDir(
+    test::PerThreadDBPath("db_log_test_db"));
 const std::string AutoRollLoggerTest::kLogFile(
     test::PerThreadDBPath("db_log_test") + "/LOG");
 Env* AutoRollLoggerTest::default_env = Env::Default();
@@ -371,7 +384,7 @@ TEST_F(AutoRollLoggerTest, CreateLoggerFromOptions) {
     options.log_file_time_to_roll = 2;
     options.keep_log_file_num = kFileNum;
     options.db_log_dir = kTestDir;
-    ASSERT_OK(CreateLoggerFromOptions("/dummy/db/name", options, &logger));
+    ASSERT_OK(CreateLoggerFromOptions(kTestDbDir, options, &logger));
     auto_roll_logger = dynamic_cast<AutoRollLogger*>(logger.get());
 
     // Roll the log 4 times, and it will trim to 3 files.
@@ -387,7 +400,7 @@ TEST_F(AutoRollLoggerTest, CreateLoggerFromOptions) {
     std::vector<std::string> files = GetLogFiles();
     ASSERT_EQ(kFileNum, files.size());
     for (const auto& f : files) {
-      ASSERT_TRUE(f.find("dummy") != std::string::npos);
+      ASSERT_TRUE(f.find("db_log_test_db") != std::string::npos);
     }
 
     // Cleaning up those files.
diff --git a/memory/arena.cc b/memory/arena.cc
index bcdad5c76..10b8969b4 100644
--- a/memory/arena.cc
+++ b/memory/arena.cc
@@ -163,7 +163,6 @@ char* Arena::AllocateAligned(size_t bytes, size_t huge_page_size,
 #ifdef MAP_HUGETLB
   if (huge_page_size > 0 && bytes > 0) {
     // Allocate from a huge page TLB table.
-    assert(logger != nullptr);  // logger need to be passed in.
     size_t reserved_size =
         ((bytes - 1U) / huge_page_size + 1U) * huge_page_size;
     assert(reserved_size >= bytes);
diff --git a/utilities/backup/backup_engine_test.cc b/utilities/backup/backup_engine_test.cc
index 3b939c5ec..6e47af89f 100644
--- a/utilities/backup/backup_engine_test.cc
+++ b/utilities/backup/backup_engine_test.cc
@@ -3066,7 +3066,7 @@ TEST_F(BackupEngineTest, OpenBackupAsReadOnlyDB) {
   db = nullptr;
 
   // Now try opening read-write and make sure it fails, for safety.
-  ASSERT_TRUE(DB::Open(opts, name, &db).IsIOError());
+  ASSERT_TRUE(DB::Open(opts, name, &db).IsAborted());
 }
 
 TEST_F(BackupEngineTest, ProgressCallbackDuringBackup) {