diff --git a/src/base/pegasus_const.cpp b/src/base/pegasus_const.cpp
index c93c0baf58..2a788cd24d 100644
--- a/src/base/pegasus_const.cpp
+++ b/src/base/pegasus_const.cpp
@@ -19,6 +19,12 @@
 
 #include "pegasus_const.h"
 
+#include <rocksdb/options.h>
+#include <stddef.h>
+#include <stdint.h>
+
+#include "utils/string_conv.h"
+
 namespace pegasus {
 
 // should be same with items in dsn::backup_restore_constant
@@ -101,4 +107,47 @@ const std::string USER_SPECIFIED_COMPACTION("user_specified_compaction");
 const std::string READ_SIZE_THROTTLING("replica.read_throttling_by_size");
 
 const std::string ROCKSDB_ALLOW_INGEST_BEHIND("rocksdb.allow_ingest_behind");
+
+const std::string ROCKSDB_WRITE_BUFFER_SIZE("rocksdb.write_buffer_size");
+
+const std::string ROCKSDB_NUM_LEVELS("rocksdb.num_levels");
+
+const std::set<std::string> ROCKSDB_DYNAMIC_OPTIONS = {
+    ROCKSDB_WRITE_BUFFER_SIZE,
+};
+const std::set<std::string> ROCKSDB_STATIC_OPTIONS = {
+    ROCKSDB_NUM_LEVELS,
+};
+
+const std::unordered_map<std::string, cf_opts_setter> cf_opts_setters = {
+    {ROCKSDB_WRITE_BUFFER_SIZE,
+     [](const std::string &str, rocksdb::ColumnFamilyOptions &option) -> bool {
+         uint64_t val = 0;
+         if (!dsn::buf2uint64(str, val)) {
+             return false;
+         }
+         option.write_buffer_size = static_cast<size_t>(val);
+         return true;
+     }},
+    {ROCKSDB_NUM_LEVELS,
+     [](const std::string &str, rocksdb::ColumnFamilyOptions &option) -> bool {
+         int32_t val = 0;
+         if (!dsn::buf2int32(str, val)) {
+             return false;
+         }
+         option.num_levels = val;
+         return true;
+     }},
+};
+
+const std::unordered_map<std::string, cf_opts_getter> cf_opts_getters = {
+    {ROCKSDB_WRITE_BUFFER_SIZE,
+     [](const rocksdb::ColumnFamilyOptions &option, /*out*/ std::string &str) {
+         str = std::to_string(option.write_buffer_size);
+     }},
+    {ROCKSDB_NUM_LEVELS,
+     [](const rocksdb::ColumnFamilyOptions &option, /*out*/ std::string &str) {
+         str = std::to_string(option.num_levels);
+     }},
+};
 } // namespace pegasus
diff --git a/src/base/pegasus_const.h b/src/base/pegasus_const.h
index c2a47ce519..e9326dfaa9 100644
--- a/src/base/pegasus_const.h
+++ b/src/base/pegasus_const.h
@@ -19,7 +19,14 @@
 
 #pragma once
 
+#include <functional>
+#include <set>
 #include <string>
+#include <unordered_map>
+
+namespace rocksdb {
+struct ColumnFamilyOptions;
+} // namespace rocksdb
 
 namespace pegasus {
 
@@ -72,4 +79,19 @@ extern const std::string USER_SPECIFIED_COMPACTION;
 extern const std::string READ_SIZE_THROTTLING;
 
 extern const std::string ROCKSDB_ALLOW_INGEST_BEHIND;
+
+extern const std::string ROCKSDB_WRITE_BUFFER_SIZE;
+
+extern const std::string ROCKSDB_NUM_LEVELS;
+
+extern const std::set<std::string> ROCKSDB_DYNAMIC_OPTIONS;
+
+extern const std::set<std::string> ROCKSDB_STATIC_OPTIONS;
+
+using cf_opts_setter = std::function<bool(const std::string &, rocksdb::ColumnFamilyOptions &)>;
+extern const std::unordered_map<std::string, cf_opts_setter> cf_opts_setters;
+
+using cf_opts_getter =
+    std::function<void(const rocksdb::ColumnFamilyOptions &, /*out*/ std::string &)>;
+extern const std::unordered_map<std::string, cf_opts_getter> cf_opts_getters;
 } // namespace pegasus
diff --git a/src/common/replica_envs.h b/src/common/replica_envs.h
index 4db367a7fa..1db2e5399f 100644
--- a/src/common/replica_envs.h
+++ b/src/common/replica_envs.h
@@ -28,6 +28,7 @@
 
 #include <cstdint>
 #include <string>
+#include <set>
 
 namespace dsn {
 namespace replication {
@@ -64,6 +65,11 @@ class replica_envs
     static const std::string USER_SPECIFIED_COMPACTION;
     static const std::string ROCKSDB_ALLOW_INGEST_BEHIND;
     static const std::string UPDATE_MAX_REPLICA_COUNT;
+    static const std::string ROCKSDB_WRITE_BUFFER_SIZE;
+    static const std::string ROCKSDB_NUM_LEVELS;
+
+    static const std::set<std::string> ROCKSDB_DYNAMIC_OPTIONS;
+    static const std::set<std::string> ROCKSDB_STATIC_OPTIONS;
 };
 
 } // namespace replication
diff --git a/src/common/replication_common.cpp b/src/common/replication_common.cpp
index 63617797ac..1afe9d49fe 100644
--- a/src/common/replication_common.cpp
+++ b/src/common/replication_common.cpp
@@ -29,6 +29,7 @@
 #include <algorithm>
 #include <fstream>
 #include <memory>
+#include <set>
 
 #include "common/gpid.h"
 #include "common/replica_envs.h"
@@ -394,6 +395,14 @@ const std::string replica_envs::USER_SPECIFIED_COMPACTION("user_specified_compac
 const std::string replica_envs::BACKUP_REQUEST_QPS_THROTTLING("replica.backup_request_throttling");
 const std::string replica_envs::ROCKSDB_ALLOW_INGEST_BEHIND("rocksdb.allow_ingest_behind");
 const std::string replica_envs::UPDATE_MAX_REPLICA_COUNT("max_replica_count.update");
+const std::string replica_envs::ROCKSDB_WRITE_BUFFER_SIZE("rocksdb.write_buffer_size");
+const std::string replica_envs::ROCKSDB_NUM_LEVELS("rocksdb.num_levels");
 
+const std::set<std::string> replica_envs::ROCKSDB_DYNAMIC_OPTIONS = {
+    replica_envs::ROCKSDB_WRITE_BUFFER_SIZE,
+};
+const std::set<std::string> replica_envs::ROCKSDB_STATIC_OPTIONS = {
+    replica_envs::ROCKSDB_NUM_LEVELS,
+};
 } // namespace replication
 } // namespace dsn
diff --git a/src/meta/app_env_validator.cpp b/src/meta/app_env_validator.cpp
index 41f9c299ce..7f197f5a0f 100644
--- a/src/meta/app_env_validator.cpp
+++ b/src/meta/app_env_validator.cpp
@@ -20,6 +20,7 @@
 #include <fmt/core.h>
 #include <stdint.h>
 #include <memory>
+#include <set>
 #include <utility>
 #include <vector>
 
@@ -31,6 +32,26 @@
 
 namespace dsn {
 namespace replication {
+bool validate_app_envs(const std::map<std::string, std::string> &envs)
+{
+    // only check rocksdb app envs currently
+
+    for (const auto &it : envs) {
+        if (replica_envs::ROCKSDB_STATIC_OPTIONS.find(it.first) ==
+                replica_envs::ROCKSDB_STATIC_OPTIONS.end() &&
+            replica_envs::ROCKSDB_DYNAMIC_OPTIONS.find(it.first) ==
+                replica_envs::ROCKSDB_DYNAMIC_OPTIONS.end()) {
+            continue;
+        }
+        std::string hint_message;
+        if (!validate_app_env(it.first, it.second, hint_message)) {
+            LOG_WARNING(
+                "app env {}={} is invaild, hint_message:{}", it.first, it.second, hint_message);
+            return false;
+        }
+    }
+    return true;
+}
 
 bool validate_app_env(const std::string &env_name,
                       const std::string &env_value,
@@ -151,6 +172,36 @@ bool check_bool_value(const std::string &env_value, std::string &hint_message)
     return true;
 }
 
+bool check_rocksdb_write_buffer_size(const std::string &env_value, std::string &hint_message)
+{
+    uint64_t val = 0;
+
+    if (!dsn::buf2uint64(env_value, val)) {
+        hint_message = fmt::format("rocksdb.write_buffer_size cannot set this val: {}", env_value);
+        return false;
+    }
+    if (val < (16 << 20) || val > (512 << 20)) {
+        hint_message =
+            fmt::format("rocksdb.write_buffer_size suggest set val in range [16777216, 536870912]");
+        return false;
+    }
+    return true;
+}
+bool check_rocksdb_num_levels(const std::string &env_value, std::string &hint_message)
+{
+    int32_t val = 0;
+
+    if (!dsn::buf2int32(env_value, val)) {
+        hint_message = fmt::format("rocksdb.num_levels cannot set this val: {}", env_value);
+        return false;
+    }
+    if (val < 1 || val > 10) {
+        hint_message = fmt::format("rocksdb.num_levels suggest set val in range [1 , 10]");
+        return false;
+    }
+    return true;
+}
+
 bool app_env_validator::validate_app_env(const std::string &env_name,
                                          const std::string &env_value,
                                          std::string &hint_message)
@@ -198,6 +249,10 @@ void app_env_validator::register_all_validators()
          std::bind(&check_bool_value, std::placeholders::_1, std::placeholders::_2)},
         {replica_envs::DENY_CLIENT_REQUEST,
          std::bind(&check_deny_client, std::placeholders::_1, std::placeholders::_2)},
+        {replica_envs::ROCKSDB_WRITE_BUFFER_SIZE,
+         std::bind(&check_rocksdb_write_buffer_size, std::placeholders::_1, std::placeholders::_2)},
+        {replica_envs::ROCKSDB_NUM_LEVELS,
+         std::bind(&check_rocksdb_num_levels, std::placeholders::_1, std::placeholders::_2)},
         // TODO(zhaoliwei): not implemented
         {replica_envs::BUSINESS_INFO, nullptr},
         {replica_envs::TABLE_LEVEL_DEFAULT_TTL, nullptr},
@@ -213,7 +268,8 @@ void app_env_validator::register_all_validators()
         {replica_envs::MANUAL_COMPACT_PERIODIC_TARGET_LEVEL, nullptr},
         {replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION, nullptr},
         {replica_envs::REPLICA_ACCESS_CONTROLLER_ALLOWED_USERS, nullptr},
-        {replica_envs::REPLICA_ACCESS_CONTROLLER_RANGER_POLICIES, nullptr}};
+        {replica_envs::REPLICA_ACCESS_CONTROLLER_RANGER_POLICIES, nullptr},
+    };
 }
 
 } // namespace replication
diff --git a/src/meta/app_env_validator.h b/src/meta/app_env_validator.h
index c401c39598..d963df0513 100644
--- a/src/meta/app_env_validator.h
+++ b/src/meta/app_env_validator.h
@@ -25,6 +25,8 @@
 namespace dsn {
 namespace replication {
 
+bool validate_app_envs(const std::map<std::string, std::string> &envs);
+
 bool validate_app_env(const std::string &env_name,
                       const std::string &env_value,
                       std::string &hint_message);
diff --git a/src/meta/server_state.cpp b/src/meta/server_state.cpp
index ecf66a92bc..2b6c6d9108 100644
--- a/src/meta/server_state.cpp
+++ b/src/meta/server_state.cpp
@@ -1151,6 +1151,9 @@ void server_state::create_app(dsn::message_ex *msg)
                !validate_target_max_replica_count(request.options.replica_count)) {
         response.err = ERR_INVALID_PARAMETERS;
         will_create_app = false;
+    } else if (!validate_app_envs(request.options.envs)) {
+        response.err = ERR_INVALID_PARAMETERS;
+        will_create_app = false;
     } else {
         zauto_write_lock l(_lock);
         app = get_app(request.app_name);
diff --git a/src/meta/test/meta_app_envs_test.cpp b/src/meta/test/meta_app_envs_test.cpp
index 255930320d..d5dc21edac 100644
--- a/src/meta/test/meta_app_envs_test.cpp
+++ b/src/meta/test/meta_app_envs_test.cpp
@@ -29,6 +29,7 @@
 #include <gtest/gtest.h>
 #include <map>
 #include <memory>
+#include <set>
 #include <string>
 
 #include "common/replica_envs.h"
@@ -142,6 +143,17 @@ TEST_F(meta_app_envs_test, update_app_envs_test)
         {replica_envs::MANUAL_COMPACT_ONCE_TARGET_LEVEL, "80", ERR_OK, "", "80"},
         {replica_envs::MANUAL_COMPACT_PERIODIC_TRIGGER_TIME, "90", ERR_OK, "", "90"},
         {replica_envs::MANUAL_COMPACT_PERIODIC_TARGET_LEVEL, "100", ERR_OK, "", "100"},
+        {replica_envs::ROCKSDB_WRITE_BUFFER_SIZE,
+         "100",
+         ERR_INVALID_PARAMETERS,
+         "rocksdb.write_buffer_size suggest set val in range [16777216, 536870912]",
+         "67108864"},
+        {replica_envs::ROCKSDB_WRITE_BUFFER_SIZE,
+         "636870912",
+         ERR_INVALID_PARAMETERS,
+         "rocksdb.write_buffer_size suggest set val in range [16777216, 536870912]",
+         "536870912"},
+        {replica_envs::ROCKSDB_WRITE_BUFFER_SIZE, "67108864", ERR_OK, "", "67108864"},
         {replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION,
          "200",
          ERR_OK,
@@ -192,6 +204,18 @@ TEST_F(meta_app_envs_test, update_app_envs_test)
             ASSERT_EQ(app->envs.at(test.env_key), test.expect_value);
         }
     }
+
+    {
+        // Make sure all rocksdb options of ROCKSDB_DYNAMIC_OPTIONS are tested.
+        // Hint: Mainly verify the update_rocksdb_dynamic_options function.
+        std::map<std::string, std::string> all_test_envs;
+        for (const auto &test : tests) {
+            all_test_envs[test.env_key] = test.env_value;
+        }
+        for (const auto &option : replica_envs::ROCKSDB_DYNAMIC_OPTIONS) {
+            ASSERT_TRUE(all_test_envs.find(option) != all_test_envs.end());
+        }
+    }
 }
 
 } // namespace replication
diff --git a/src/meta/test/meta_app_operation_test.cpp b/src/meta/test/meta_app_operation_test.cpp
index c3ddafa72c..e0afc1baa5 100644
--- a/src/meta/test/meta_app_operation_test.cpp
+++ b/src/meta/test/meta_app_operation_test.cpp
@@ -24,6 +24,7 @@
 #include <iostream>
 #include <map>
 #include <memory>
+#include <set>
 #include <string>
 #include <utility>
 #include <vector>
@@ -68,7 +69,8 @@ class meta_app_operation_test : public meta_test_base
     error_code create_app_test(int32_t partition_count,
                                int32_t replica_count,
                                bool success_if_exist,
-                               const std::string &app_name)
+                               const std::string &app_name,
+                               const std::map<std::string, std::string> &envs = {})
     {
         configuration_create_app_request create_request;
         configuration_create_app_response create_response;
@@ -78,6 +80,7 @@ class meta_app_operation_test : public meta_test_base
         create_request.options.replica_count = replica_count;
         create_request.options.success_if_exist = success_if_exist;
         create_request.options.is_stateful = true;
+        create_request.options.envs = envs;
 
         auto result = fake_create_app(_ss.get(), create_request);
         fake_wait_rpc(result, create_response);
@@ -362,6 +365,14 @@ TEST_F(meta_app_operation_test, create_app)
     // - wrong app_status dropping
     // - create succeed with app_status dropped
     // - create succeed with success_if_exist=true
+    // - wrong rocksdb.num_levels (< 1)
+    // - wrong rocksdb.num_levels (> 10)
+    // - wrong rocksdb.num_levels (non-digital character)
+    // - create app with rocksdb.num_levels (= 5) succeed
+    // - wrong rocksdb.write_buffer_size (< (16<<20))
+    // - wrong rocksdb.write_buffer_size (> (512<<20))
+    // - wrong rocksdb.write_buffer_size (non-digital character)
+    // - create app with rocksdb.write_buffer_size (= (32<<20)) succeed
     struct create_test
     {
         std::string app_name;
@@ -373,43 +384,126 @@ TEST_F(meta_app_operation_test, create_app)
         bool success_if_exist;
         app_status::type before_status;
         error_code expected_err;
-    } tests[] = {{APP_NAME, -1, 3, 2, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 0, 3, 2, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, -1, 1, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 0, 1, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 6, 2, 4, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 7, 2, 6, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 6, 2, 5, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 5, 2, 4, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 4, 2, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 6, 2, 6, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 6, 2, 7, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME + "_1", 4, 5, 2, 5, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_2", 4, 5, 2, 6, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_3", 4, 4, 2, 4, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_4", 4, 4, 2, 6, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_5", 4, 3, 2, 4, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_6", 4, 4, 2, 5, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME, 4, 3, 2, 5, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 3, 2, 4, 5, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 3, 2, 4, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 3, 2, 2, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 3, 2, 3, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME, 4, 4, 2, 3, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
-                 {APP_NAME + "_7", 4, 3, 2, 4, 3, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME, 4, 1, 1, 0, 1, false, app_status::AS_INVALID, ERR_STATE_FREEZED},
-                 {APP_NAME, 4, 2, 2, 1, 1, false, app_status::AS_INVALID, ERR_STATE_FREEZED},
-                 {APP_NAME, 4, 3, 3, 2, 1, false, app_status::AS_INVALID, ERR_STATE_FREEZED},
-                 {APP_NAME + "_8", 4, 3, 3, 3, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_9", 4, 1, 1, 1, 1, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME + "_10", 4, 2, 1, 2, 2, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_INVALID, ERR_OK},
-                 {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_INVALID, ERR_APP_EXIST},
-                 {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_CREATING, ERR_BUSY_CREATING},
-                 {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_RECALLING, ERR_BUSY_CREATING},
-                 {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_DROPPING, ERR_BUSY_DROPPING},
-                 {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_DROPPED, ERR_OK},
-                 {APP_NAME, 4, 3, 2, 3, 3, true, app_status::AS_INVALID, ERR_OK}};
+        std::map<std::string, std::string> envs = {};
+    } tests[] = {
+        {APP_NAME, -1, 3, 2, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 0, 3, 2, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, -1, 1, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 0, 1, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 6, 2, 4, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 7, 2, 6, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 6, 2, 5, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 5, 2, 4, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 4, 2, 3, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 6, 2, 6, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 6, 2, 7, 1, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME + "_1", 4, 5, 2, 5, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_2", 4, 5, 2, 6, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_3", 4, 4, 2, 4, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_4", 4, 4, 2, 6, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_5", 4, 3, 2, 4, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_6", 4, 4, 2, 5, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME, 4, 3, 2, 5, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 3, 2, 4, 5, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 3, 2, 4, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 3, 2, 2, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 3, 2, 3, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME, 4, 4, 2, 3, 4, false, app_status::AS_INVALID, ERR_INVALID_PARAMETERS},
+        {APP_NAME + "_7", 4, 3, 2, 4, 3, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME, 4, 1, 1, 0, 1, false, app_status::AS_INVALID, ERR_STATE_FREEZED},
+        {APP_NAME, 4, 2, 2, 1, 1, false, app_status::AS_INVALID, ERR_STATE_FREEZED},
+        {APP_NAME, 4, 3, 3, 2, 1, false, app_status::AS_INVALID, ERR_STATE_FREEZED},
+        {APP_NAME + "_8", 4, 3, 3, 3, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_9", 4, 1, 1, 1, 1, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME + "_10", 4, 2, 1, 2, 2, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_INVALID, ERR_APP_EXIST},
+        {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_CREATING, ERR_BUSY_CREATING},
+        {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_RECALLING, ERR_BUSY_CREATING},
+        {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_DROPPING, ERR_BUSY_DROPPING},
+        {APP_NAME, 4, 3, 2, 3, 3, false, app_status::AS_DROPPED, ERR_OK},
+        {APP_NAME, 4, 3, 2, 3, 3, true, app_status::AS_INVALID, ERR_OK},
+        {APP_NAME,
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_INVALID_PARAMETERS,
+         {{"rocksdb.num_levels", "0"}}},
+        {APP_NAME,
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_INVALID_PARAMETERS,
+         {{"rocksdb.num_levels", "11"}}},
+        {APP_NAME + "_11",
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_INVALID_PARAMETERS,
+         {{"rocksdb.num_levels", "5i"}}},
+        {APP_NAME + "_11",
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_OK,
+         {{"rocksdb.num_levels", "5"}}},
+        {APP_NAME,
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_INVALID_PARAMETERS,
+         {{"rocksdb.write_buffer_size", "1000"}}},
+        {APP_NAME,
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_INVALID_PARAMETERS,
+         {{"rocksdb.write_buffer_size", "1073741824"}}},
+        {APP_NAME,
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_INVALID_PARAMETERS,
+         {{"rocksdb.write_buffer_size", "n33554432"}}},
+        {APP_NAME + "_12",
+         4,
+         3,
+         2,
+         3,
+         3,
+         false,
+         app_status::AS_INVALID,
+         ERR_OK,
+         {{"rocksdb.write_buffer_size", "33554432"}}},
+    };
 
     clear_nodes();
 
@@ -453,13 +547,30 @@ TEST_F(meta_app_operation_test, create_app)
         } else if (test.before_status != app_status::AS_INVALID) {
             update_app_status(test.before_status);
         }
-        auto err = create_app_test(
-            test.partition_count, test.replica_count, test.success_if_exist, test.app_name);
+        auto err = create_app_test(test.partition_count,
+                                   test.replica_count,
+                                   test.success_if_exist,
+                                   test.app_name,
+                                   test.envs);
         ASSERT_EQ(err, test.expected_err);
 
         _ms->set_node_state(nodes, true);
     }
 
+    {
+        // Make sure all rocksdb options of ROCKSDB_DYNAMIC_OPTIONS and ROCKSDB_STATIC_OPTIONS are
+        // tested. Hint: Mainly verify the validate_app_envs function.
+        std::map<std::string, std::string> all_test_envs;
+        for (const auto &test : tests) {
+            all_test_envs.insert(test.envs.begin(), test.envs.end());
+        }
+        for (const auto &option : replica_envs::ROCKSDB_DYNAMIC_OPTIONS) {
+            ASSERT_TRUE(all_test_envs.find(option) != all_test_envs.end());
+        }
+        for (const auto &option : replica_envs::ROCKSDB_STATIC_OPTIONS) {
+            ASSERT_TRUE(all_test_envs.find(option) != all_test_envs.end());
+        }
+    }
     // set FLAGS_min_allowed_replica_count successfully
     res = update_flag("min_allowed_replica_count", "2");
     ASSERT_TRUE(res.is_ok());
diff --git a/src/server/pegasus_server_impl.cpp b/src/server/pegasus_server_impl.cpp
index e6e9d48edc..998b4873f7 100644
--- a/src/server/pegasus_server_impl.cpp
+++ b/src/server/pegasus_server_impl.cpp
@@ -38,10 +38,12 @@
 #include <unistd.h> // IWYU pragma: keep
 #include <algorithm>
 #include <cstdint>
+#include <functional>
 #include <limits>
 #include <list>
 #include <mutex>
 #include <ostream>
+#include <set>
 
 #include "base/pegasus_key_schema.h"
 #include "base/pegasus_utils.h"
@@ -1663,7 +1665,7 @@ dsn::error_code pegasus_server_impl::start(int argc, char **argv)
             // We don't use `loaded_data_cf_opts` directly because pointer-typed options will
             // only be initialized with default values when calling 'LoadLatestOptions', see
             // 'rocksdb/utilities/options_util.h'.
-            reset_usage_scenario_options(loaded_data_cf_opts, &_table_data_cf_opts);
+            reset_rocksdb_options(loaded_data_cf_opts, &_table_data_cf_opts);
             _db_opts.allow_ingest_behind = parse_allow_ingest_behind(envs);
         }
     } else {
@@ -2625,6 +2627,67 @@ pegasus_server_impl::get_restore_dir_from_env(const std::map<std::string, std::s
     return res;
 }
 
+void pegasus_server_impl::update_rocksdb_dynamic_options(
+    const std::map<std::string, std::string> &envs)
+{
+    if (envs.empty()) {
+        return;
+    }
+
+    std::unordered_map<std::string, std::string> new_options;
+    for (const auto &option : ROCKSDB_DYNAMIC_OPTIONS) {
+        const auto &find = envs.find(option);
+        if (find == envs.end()) {
+            continue;
+        }
+
+        std::vector<std::string> args;
+        // split_args example: Parse "write_buffer_size" from "rocksdb.write_buffer_size"
+        dsn::utils::split_args(option.c_str(), args, '.');
+        CHECK_EQ(args.size(), 2);
+        new_options[args[1]] = find->second;
+    }
+
+    // doing set option
+    if (!new_options.empty() && set_options(new_options)) {
+        LOG_INFO("Set rocksdb dynamic options success");
+    }
+}
+
+void pegasus_server_impl::set_rocksdb_options_before_creating(
+    const std::map<std::string, std::string> &envs)
+{
+    if (envs.empty()) {
+        return;
+    }
+
+    for (const auto &option : pegasus::ROCKSDB_STATIC_OPTIONS) {
+        const auto &find = envs.find(option);
+        if (find == envs.end()) {
+            continue;
+        }
+
+        const auto &setter = cf_opts_setters.find(option);
+        CHECK_TRUE(setter != cf_opts_setters.end());
+        if (setter->second(find->second, _data_cf_opts)) {
+            LOG_INFO_PREFIX("Set {} \"{}\" succeed", find->first, find->second);
+        }
+    }
+
+    for (const auto &option : pegasus::ROCKSDB_DYNAMIC_OPTIONS) {
+        const auto &find = envs.find(option);
+        if (find == envs.end()) {
+            continue;
+        }
+
+        const auto &setter = cf_opts_setters.find(option);
+        CHECK_TRUE(setter != cf_opts_setters.end());
+        if (setter->second(find->second, _data_cf_opts)) {
+            LOG_INFO_PREFIX("Set {} \"{}\" succeed", find->first, find->second);
+        }
+    }
+}
+
 void pegasus_server_impl::update_app_envs(const std::map<std::string, std::string> &envs)
 {
     update_usage_scenario(envs);
@@ -2637,6 +2700,7 @@ void pegasus_server_impl::update_app_envs(const std::map<std::string, std::strin
     _manual_compact_svc.start_manual_compact_if_needed(envs);
 
     update_throttling_controller(envs);
+    update_rocksdb_dynamic_options(envs);
 }
 
 void pegasus_server_impl::update_app_envs_before_open_db(
@@ -2650,11 +2714,36 @@ void pegasus_server_impl::update_app_envs_before_open_db(
     update_validate_partition_hash(envs);
     update_user_specified_compaction(envs);
     _manual_compact_svc.start_manual_compact_if_needed(envs);
+    set_rocksdb_options_before_creating(envs);
 }
 
 void pegasus_server_impl::query_app_envs(/*out*/ std::map<std::string, std::string> &envs)
 {
     envs[ROCKSDB_ENV_USAGE_SCENARIO_KEY] = _usage_scenario;
+    // write_buffer_size involves random values (refer to pegasus_server_impl::set_usage_scenario),
+    // so it can only be taken from _data_cf_opts
+    envs[ROCKSDB_WRITE_BUFFER_SIZE] = std::to_string(_data_cf_opts.write_buffer_size);
+
+    // Get Data ColumnFamilyOptions directly from _data_cf
+    rocksdb::ColumnFamilyDescriptor desc;
+    CHECK_TRUE(_data_cf->GetDescriptor(&desc).ok());
+    for (const auto &option : pegasus::ROCKSDB_STATIC_OPTIONS) {
+        auto getter = cf_opts_getters.find(option);
+        CHECK_TRUE(getter != cf_opts_getters.end());
+        std::string option_val;
+        getter->second(desc.options, option_val);
+        envs[option] = option_val;
+    }
+    for (const auto &option : pegasus::ROCKSDB_DYNAMIC_OPTIONS) {
+        if (option.compare(ROCKSDB_WRITE_BUFFER_SIZE) == 0) {
+            continue;
+        }
+        auto getter = cf_opts_getters.find(option);
+        CHECK_TRUE(getter != cf_opts_getters.end());
+        std::string option_val;
+        getter->second(desc.options, option_val);
+        envs[option] = option_val;
+    }
 }
 
 void pegasus_server_impl::update_usage_scenario(const std::map<std::string, std::string> &envs)
@@ -3038,6 +3127,23 @@ bool pegasus_server_impl::set_usage_scenario(const std::string &usage_scenario)
     }
 }
 
+void pegasus_server_impl::reset_rocksdb_options(const rocksdb::ColumnFamilyOptions &base_opts,
+                                                rocksdb::ColumnFamilyOptions *target_opts)
+{
+    LOG_INFO_PREFIX("Reset rocksdb envs options");
+    // Reset rocksdb option includes two aspects:
+    // 1. Set usage_scenario related rocksdb options
+    // 2. Rocksdb option set in app envs, consists of ROCKSDB_DYNAMIC_OPTIONS and
+    // ROCKSDB_STATIC_OPTIONS
+
+    // aspect 1:
+    reset_usage_scenario_options(base_opts, target_opts);
+
+    // aspect 2:
+    target_opts->num_levels = base_opts.num_levels;
+    target_opts->write_buffer_size = base_opts.write_buffer_size;
+}
+
 void pegasus_server_impl::reset_usage_scenario_options(
     const rocksdb::ColumnFamilyOptions &base_opts, rocksdb::ColumnFamilyOptions *target_opts)
 {
diff --git a/src/server/pegasus_server_impl.h b/src/server/pegasus_server_impl.h
index d156acdce6..c9c5b90ef6 100644
--- a/src/server/pegasus_server_impl.h
+++ b/src/server/pegasus_server_impl.h
@@ -22,6 +22,7 @@
 #include <gtest/gtest_prod.h>
 #include <rocksdb/options.h>
 #include <rocksdb/slice.h>
+#include <rocksdb/table.h>
 #include <rrdb/rrdb_types.h>
 #include <stdint.h>
 #include <atomic>
@@ -328,6 +329,10 @@ class pegasus_server_impl : public pegasus_read_service
 
     void update_user_specified_compaction(const std::map<std::string, std::string> &envs);
 
+    void update_rocksdb_dynamic_options(const std::map<std::string, std::string> &envs);
+
+    void set_rocksdb_options_before_creating(const std::map<std::string, std::string> &envs);
+
     void update_throttling_controller(const std::map<std::string, std::string> &envs);
 
     bool parse_allow_ingest_behind(const std::map<std::string, std::string> &envs);
@@ -359,6 +364,9 @@ class pegasus_server_impl : public pegasus_read_service
     void reset_usage_scenario_options(const rocksdb::ColumnFamilyOptions &base_opts,
                                       rocksdb::ColumnFamilyOptions *target_opts);
 
+    void reset_rocksdb_options(const rocksdb::ColumnFamilyOptions &base_opts,
+                               rocksdb::ColumnFamilyOptions *target_opts);
+
     // return true if successfully set
     bool set_options(const std::unordered_map<std::string, std::string> &new_options);
 
@@ -468,6 +476,7 @@ class pegasus_server_impl : public pegasus_read_service
     // Dynamically calculate the value of current data_cf option according to the conf module file
     // and usage scenario
     rocksdb::ColumnFamilyOptions _table_data_cf_opts;
+    rocksdb::BlockBasedTableOptions _tbl_opts;
     rocksdb::ColumnFamilyOptions _meta_cf_opts;
     rocksdb::ReadOptions _data_cf_rd_opts;
     std::string _usage_scenario;
diff --git a/src/server/pegasus_server_impl_init.cpp b/src/server/pegasus_server_impl_init.cpp
index 896f8ab945..c457ebd9fa 100644
--- a/src/server/pegasus_server_impl_init.cpp
+++ b/src/server/pegasus_server_impl_init.cpp
@@ -137,7 +137,7 @@ DSN_DEFINE_bool(pegasus.server,
 DSN_DEFINE_bool(pegasus.server,
                 rocksdb_disable_table_block_cache,
                 false,
-                "rocksdb tbl_opts.no_block_cache");
+                "rocksdb _tbl_opts.no_block_cache");
 DSN_DEFINE_bool(pegasus.server,
                 rocksdb_enable_write_buffer_manager,
                 false,
@@ -466,12 +466,11 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
     CHECK(parse_compression_types("none", _meta_cf_opts.compression_per_level),
           "parse rocksdb_compression_type failed.");
 
-    rocksdb::BlockBasedTableOptions tbl_opts;
-    tbl_opts.read_amp_bytes_per_bit = FLAGS_read_amp_bytes_per_bit;
+    _tbl_opts.read_amp_bytes_per_bit = FLAGS_read_amp_bytes_per_bit;
 
     if (FLAGS_rocksdb_disable_table_block_cache) {
-        tbl_opts.no_block_cache = true;
-        tbl_opts.block_restart_interval = 4;
+        _tbl_opts.no_block_cache = true;
+        _tbl_opts.block_restart_interval = 4;
     } else {
         // If block cache is enabled, all replicas on this server will share the same block cache
         // object. It's convenient to control the total memory used by this server, and the LRU
@@ -484,7 +483,7 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
         });
 
         // every replica has the same block cache
-        tbl_opts.block_cache = _s_block_cache;
+        _tbl_opts.block_cache = _s_block_cache;
     }
 
     // FLAGS_rocksdb_limiter_max_write_megabytes_per_sec <= 0 means close the rate limit.
@@ -520,7 +519,7 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
                             FLAGS_rocksdb_total_size_across_write_buffer);
             _s_write_buffer_manager = std::make_shared<rocksdb::WriteBufferManager>(
                 static_cast<size_t>(FLAGS_rocksdb_total_size_across_write_buffer),
-                tbl_opts.block_cache);
+                _tbl_opts.block_cache);
         });
         _db_opts.write_buffer_manager = _s_write_buffer_manager;
     }
@@ -541,33 +540,33 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
     CHECK(index_type_item != INDEX_TYPE_STRING_MAP.end(),
           "[pegasus.server]rocksdb_index_type should be one among binary_search, "
           "hash_search, two_level_index_search or binary_search_with_first_key.");
-    tbl_opts.index_type = index_type_item->second;
+    _tbl_opts.index_type = index_type_item->second;
     LOG_INFO_PREFIX("rocksdb_index_type = {}", FLAGS_rocksdb_index_type);
 
-    tbl_opts.partition_filters = FLAGS_rocksdb_partition_filters;
+    _tbl_opts.partition_filters = FLAGS_rocksdb_partition_filters;
     // TODO(yingchun): clean up these useless log ?
-    LOG_INFO_PREFIX("rocksdb_partition_filters = {}", tbl_opts.partition_filters);
+    LOG_INFO_PREFIX("rocksdb_partition_filters = {}", _tbl_opts.partition_filters);
 
-    tbl_opts.metadata_block_size = FLAGS_rocksdb_metadata_block_size;
-    LOG_INFO_PREFIX("rocksdb_metadata_block_size = {}", tbl_opts.metadata_block_size);
+    _tbl_opts.metadata_block_size = FLAGS_rocksdb_metadata_block_size;
+    LOG_INFO_PREFIX("rocksdb_metadata_block_size = {}", _tbl_opts.metadata_block_size);
 
-    tbl_opts.cache_index_and_filter_blocks = FLAGS_rocksdb_cache_index_and_filter_blocks;
+    _tbl_opts.cache_index_and_filter_blocks = FLAGS_rocksdb_cache_index_and_filter_blocks;
     LOG_INFO_PREFIX("rocksdb_cache_index_and_filter_blocks = {}",
-                    tbl_opts.cache_index_and_filter_blocks);
+                    _tbl_opts.cache_index_and_filter_blocks);
 
-    tbl_opts.pin_top_level_index_and_filter = FLAGS_rocksdb_pin_top_level_index_and_filter;
+    _tbl_opts.pin_top_level_index_and_filter = FLAGS_rocksdb_pin_top_level_index_and_filter;
     LOG_INFO_PREFIX("rocksdb_pin_top_level_index_and_filter = {}",
-                    tbl_opts.pin_top_level_index_and_filter);
+                    _tbl_opts.pin_top_level_index_and_filter);
 
-    tbl_opts.cache_index_and_filter_blocks_with_high_priority =
+    _tbl_opts.cache_index_and_filter_blocks_with_high_priority =
         FLAGS_rocksdb_cache_index_and_filter_blocks_with_high_priority;
     LOG_INFO_PREFIX("rocksdb_cache_index_and_filter_blocks_with_high_priority = {}",
-                    tbl_opts.cache_index_and_filter_blocks_with_high_priority);
+                    _tbl_opts.cache_index_and_filter_blocks_with_high_priority);
 
-    tbl_opts.pin_l0_filter_and_index_blocks_in_cache =
+    _tbl_opts.pin_l0_filter_and_index_blocks_in_cache =
         FLAGS_rocksdb_pin_l0_filter_and_index_blocks_in_cache;
     LOG_INFO_PREFIX("rocksdb_pin_l0_filter_and_index_blocks_in_cache = {}",
-                    tbl_opts.pin_l0_filter_and_index_blocks_in_cache);
+                    _tbl_opts.pin_l0_filter_and_index_blocks_in_cache);
 
     // Bloom filter configurations.
     if (!FLAGS_rocksdb_disable_bloom_filter) {
@@ -584,8 +583,8 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
         //                                 50         |      0.225453      |      ~0.00003
         // Recommend using no more than three decimal digits after the decimal point, as in 6.667.
         // More details: https://github.com/facebook/rocksdb/wiki/RocksDB-Bloom-Filter
-        tbl_opts.format_version = FLAGS_rocksdb_format_version;
-        tbl_opts.filter_policy.reset(
+        _tbl_opts.format_version = FLAGS_rocksdb_format_version;
+        _tbl_opts.filter_policy.reset(
             rocksdb::NewBloomFilterPolicy(FLAGS_rocksdb_bloom_filter_bits_per_key, false));
 
         if (dsn::utils::equals(FLAGS_rocksdb_filter_type, "prefix")) {
@@ -596,8 +595,8 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
         }
     }
 
-    _data_cf_opts.table_factory.reset(NewBlockBasedTableFactory(tbl_opts));
-    _meta_cf_opts.table_factory.reset(NewBlockBasedTableFactory(tbl_opts));
+    _data_cf_opts.table_factory.reset(NewBlockBasedTableFactory(_tbl_opts));
+    _meta_cf_opts.table_factory.reset(NewBlockBasedTableFactory(_tbl_opts));
 
     _key_ttl_compaction_filter_factory = std::make_shared<KeyWithTTLCompactionFilterFactory>();
     _data_cf_opts.compaction_filter_factory = _key_ttl_compaction_filter_factory;
diff --git a/src/server/test/pegasus_server_impl_test.cpp b/src/server/test/pegasus_server_impl_test.cpp
index 1363d8054b..9776571ae9 100644
--- a/src/server/test/pegasus_server_impl_test.cpp
+++ b/src/server/test/pegasus_server_impl_test.cpp
@@ -29,7 +29,9 @@
 #include <stdint.h>
 #include <map>
 #include <memory>
+#include <set>
 #include <string>
+#include <utility>
 
 #include "pegasus_const.h"
 #include "pegasus_server_test_base.h"
@@ -95,6 +97,50 @@ class pegasus_server_impl_test : public pegasus_server_test_base
             ASSERT_EQ(before_count + test.expect_perf_counter_incr, after_count);
         }
     }
+
+    void test_open_db_with_rocksdb_envs(bool is_restart)
+    {
+        struct create_test
+        {
+            std::string env_key;
+            std::string env_value;
+            std::string expect_value;
+        } tests[] = {
+            {"rocksdb.num_levels", "5", "5"}, {"rocksdb.write_buffer_size", "33554432", "33554432"},
+        };
+
+        std::map<std::string, std::string> all_test_envs;
+        {
+            // Make sure all rocksdb options of ROCKSDB_DYNAMIC_OPTIONS and ROCKSDB_STATIC_OPTIONS
+            // are tested.
+            for (const auto &test : tests) {
+                all_test_envs[test.env_key] = test.env_value;
+            }
+            for (const auto &option : pegasus::ROCKSDB_DYNAMIC_OPTIONS) {
+                ASSERT_TRUE(all_test_envs.find(option) != all_test_envs.end());
+            }
+            for (const auto &option : pegasus::ROCKSDB_STATIC_OPTIONS) {
+                ASSERT_TRUE(all_test_envs.find(option) != all_test_envs.end());
+            }
+        }
+
+        start(all_test_envs);
+        if (is_restart) {
+            _server->stop(false);
+            start();
+        }
+
+        std::map<std::string, std::string> query_envs;
+        _server->query_app_envs(query_envs);
+        for (const auto &test : tests) {
+            const auto &iter = query_envs.find(test.env_key);
+            if (iter != query_envs.end()) {
+                ASSERT_EQ(iter->second, test.expect_value);
+            } else {
+                ASSERT_TRUE(false) << fmt::format("query_app_envs not supported {}", test.env_key);
+            }
+        }
+    }
 };
 
 TEST_F(pegasus_server_impl_test, test_table_level_slow_query)
@@ -137,6 +183,18 @@ TEST_F(pegasus_server_impl_test, test_open_db_with_app_envs)
     ASSERT_EQ(ROCKSDB_ENV_USAGE_SCENARIO_BULK_LOAD, _server->_usage_scenario);
 }
 
+TEST_F(pegasus_server_impl_test, test_open_db_with_rocksdb_envs)
+{
+    // Hint: Verify the set_rocksdb_options_before_creating function by boolean is_restart=false.
+    test_open_db_with_rocksdb_envs(false);
+}
+
+TEST_F(pegasus_server_impl_test, test_restart_db_with_rocksdb_envs)
+{
+    // Hint: Verify the reset_rocksdb_options function by boolean is_restart=true.
+    test_open_db_with_rocksdb_envs(true);
+}
+
 TEST_F(pegasus_server_impl_test, test_stop_db_twice)
 {
     start();
diff --git a/src/utils/simple_logger.cpp b/src/utils/simple_logger.cpp
index f982670186..142c3de233 100644
--- a/src/utils/simple_logger.cpp
+++ b/src/utils/simple_logger.cpp
@@ -41,6 +41,7 @@
 #include "utils/filesystem.h"
 #include "utils/flags.h"
 #include "utils/fmt_logging.h"
+#include "utils/ports.h"
 #include "utils/process_utils.h"
 #include "utils/strings.h"
 #include "utils/time_utils.h"