diff --git a/src/meta/meta_backup_service.cpp b/src/meta/meta_backup_service.cpp
index 12bff32b2d..db2afe1d90 100644
--- a/src/meta/meta_backup_service.cpp
+++ b/src/meta/meta_backup_service.cpp
@@ -41,6 +41,7 @@
 #include "runtime/rpc/rpc_holder.h"
 #include "runtime/rpc/rpc_message.h"
 #include "runtime/rpc/serialization.h"
+#include "runtime/security/access_controller.h"
 #include "runtime/task/async_calls.h"
 #include "runtime/task/task_code.h"
 #include "server_state.h"
@@ -1248,6 +1249,7 @@ void backup_service::add_backup_policy(dsn::message_ex *msg)
     configuration_add_backup_policy_request request;
     configuration_add_backup_policy_response response;
 
+    dsn::message_ex *copied_msg = message_ex::copy_message_no_reply(*msg);
     ::dsn::unmarshall(msg, request);
     std::set<int32_t> app_ids;
     std::map<int32_t, std::string> app_names;
@@ -1280,6 +1282,20 @@ void backup_service::add_backup_policy(dsn::message_ex *msg)
                 msg->release_ref();
                 return;
             }
+            // when the Ranger ACL is enabled, access control will be checked for each table.
+            auto access_controller = _meta_svc->get_access_controller();
+            // adding multiple judgments here is to adapt to the old ACL and avoid checking again.
+            if (access_controller->is_enable_ranger_acl() &&
+                !access_controller->allowed(copied_msg, app->app_name)) {
+                response.err = ERR_ACL_DENY;
+                response.hint_message =
+                    fmt::format("not authorized to add backup policy({}) for app id: {}",
+                                request.policy_name,
+                                app_id);
+                _meta_svc->reply_data(msg, response);
+                msg->release_ref();
+                return;
+            }
             app_ids.insert(app_id);
             app_names.insert(std::make_pair(app_id, app->app_name));
         }
@@ -1510,16 +1526,24 @@ void backup_service::modify_backup_policy(configuration_modify_backup_policy_rpc
 
         for (const auto &appid : request.add_appids) {
             const auto &app = _state->get_app(appid);
+            auto access_controller = _meta_svc->get_access_controller();
             // TODO: if app is dropped, how to process
             if (app == nullptr) {
                 LOG_WARNING("{}: add app to policy failed, because invalid app({}), ignore it",
                             cur_policy.policy_name,
                             appid);
-            } else {
-                valid_app_ids_to_add.emplace_back(appid);
-                id_to_app_names.insert(std::make_pair(appid, app->app_name));
-                have_modify_policy = true;
+                continue;
             }
+            if (access_controller->is_enable_ranger_acl() &&
+                !access_controller->allowed(rpc.dsn_request(), app->app_name)) {
+                LOG_WARNING("not authorized to modify backup policy({}) for app id: {}, skip it",
+                            cur_policy.policy_name,
+                            appid);
+                continue;
+            }
+            valid_app_ids_to_add.emplace_back(appid);
+            id_to_app_names.insert(std::make_pair(appid, app->app_name));
+            have_modify_policy = true;
         }
     }
 
diff --git a/src/meta/meta_service.cpp b/src/meta/meta_service.cpp
index d2f3fd00d8..4b65d7b299 100644
--- a/src/meta/meta_service.cpp
+++ b/src/meta/meta_service.cpp
@@ -38,6 +38,7 @@
 #include "backup_types.h"
 #include "bulk_load_types.h"
 #include "common/common.h"
+#include "common/gpid.h"
 #include "common/replication.codes.h"
 #include "dsn.layer2_types.h"
 #include "duplication_types.h"
@@ -56,6 +57,7 @@
 #include "partition_split_types.h"
 #include "perf_counter/perf_counter.h"
 #include "remote_cmd/remote_command.h"
+#include "runtime/ranger/ranger_resource_policy_manager.h"
 #include "runtime/rpc/rpc_holder.h"
 #include "runtime/task/async_calls.h"
 #include "server_load_balancer.h"
@@ -124,6 +126,21 @@ DSN_DECLARE_int32(fd_grace_seconds);
 DSN_DECLARE_int32(fd_lease_seconds);
 DSN_DECLARE_string(cold_backup_root);
 
+#define CHECK_APP_ID_STATUS_AND_AUTHZ(app_id)                                                      \
+    do {                                                                                           \
+        const auto &_app_id = (app_id);                                                            \
+        const auto &_app = _state->get_app(_app_id);                                               \
+        if (!_app) {                                                                               \
+            rpc.response().err = ERR_INVALID_PARAMETERS;                                           \
+            LOG_WARNING("reject request on app_id = {}", _app_id);                                 \
+            return;                                                                                \
+        }                                                                                          \
+        const std::string &app_name = _app->app_name;                                              \
+        if (!check_status_and_authz(rpc, nullptr, app_name)) {                                     \
+            return;                                                                                \
+        }                                                                                          \
+    } while (0)
+
 meta_service::meta_service()
     : serverlet("meta_service"), _failure_detector(nullptr), _started(false), _recovering(false)
 {
@@ -151,8 +168,6 @@ meta_service::meta_service()
     _alive_nodes_count.init_app_counter(
         "eon.meta_service", "alive_nodes", COUNTER_TYPE_NUMBER, "current count of alive nodes");
 
-    _access_controller = security::create_meta_access_controller();
-
     _meta_op_status.store(meta_op_status::FREE);
 }
 
@@ -320,6 +335,11 @@ void meta_service::start_service()
         _failure_detector->register_worker(node, true);
     }
 
+    _ranger_resource_policy_manager =
+        std::make_shared<ranger::ranger_resource_policy_manager>(this);
+
+    _access_controller = security::create_meta_access_controller(_ranger_resource_policy_manager);
+
     _started = true;
     for (const dsn::rpc_address &node : _alive_set) {
         tasking::enqueue(LPC_META_STATE_HIGH,
@@ -567,8 +587,8 @@ meta_leader_state meta_service::check_leader(dsn::message_ex *req,
 // table operations
 void meta_service::on_create_app(dsn::message_ex *req)
 {
-    configuration_create_app_response response;
-    if (!check_status_with_msg(req, response)) {
+    if (!check_status_and_authz_with_reply<configuration_create_app_request,
+                                           configuration_create_app_response>(req)) {
         return;
     }
 
@@ -581,8 +601,8 @@ void meta_service::on_create_app(dsn::message_ex *req)
 
 void meta_service::on_drop_app(dsn::message_ex *req)
 {
-    configuration_drop_app_response response;
-    if (!check_status_with_msg(req, response)) {
+    if (!check_status_and_authz_with_reply<configuration_drop_app_request,
+                                           configuration_drop_app_response>(req)) {
         return;
     }
 
@@ -595,7 +615,7 @@ void meta_service::on_drop_app(dsn::message_ex *req)
 
 void meta_service::on_rename_app(configuration_rename_app_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().old_app_name)) {
         return;
     }
 
@@ -607,10 +627,34 @@ void meta_service::on_rename_app(configuration_rename_app_rpc rpc)
 
 void meta_service::on_recall_app(dsn::message_ex *req)
 {
+    configuration_recall_app_request request;
     configuration_recall_app_response response;
-    if (!check_status_with_msg(req, response)) {
+    dsn::message_ex *copied_req = message_ex::copy_message_no_reply(*req);
+    dsn::unmarshall(copied_req, request);
+    auto target_app = _state->get_app(request.app_id);
+    if (!target_app) {
+        response.err = ERR_APP_NOT_EXIST;
+        reply(req, response);
+        return;
+    }
+    const std::string &app_name = target_app->app_name;
+
+    if (!check_status_and_authz_with_reply(req, response, app_name)) {
         return;
     }
+    // check new_app_name reasonable.
+    // when the Ranger ACL is enabled, ensure that the prefix of new_app_name is consistent with
+    // old, or it is empty
+    if (_access_controller->is_enable_ranger_acl() && !request.new_app_name.empty()) {
+        std::string app_name_prefix = ranger::get_database_name_from_app_name(app_name);
+        std::string new_app_name_prefix =
+            ranger::get_database_name_from_app_name(request.new_app_name);
+        if (app_name_prefix != new_app_name_prefix) {
+            response.err = ERR_INVALID_PARAMETERS;
+            reply(req, response);
+            return;
+        }
+    }
 
     req->add_ref();
     tasking::enqueue(LPC_META_STATE_NORMAL,
@@ -621,16 +665,20 @@ void meta_service::on_recall_app(dsn::message_ex *req)
 
 void meta_service::on_list_apps(configuration_list_apps_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_leader_status(rpc)) {
         return;
     }
 
-    _state->list_apps(rpc.request(), rpc.response());
+    dsn::message_ex *msg = nullptr;
+    if (_access_controller->is_enable_ranger_acl()) {
+        msg = rpc.dsn_request();
+    }
+    _state->list_apps(rpc.request(), rpc.response(), msg);
 }
 
 void meta_service::on_list_nodes(configuration_list_nodes_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -660,7 +708,7 @@ void meta_service::on_list_nodes(configuration_list_nodes_rpc rpc)
 
 void meta_service::on_query_cluster_info(configuration_cluster_info_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -703,7 +751,7 @@ void meta_service::on_query_configuration_by_index(configuration_query_by_index_
 {
     query_cfg_response &response = rpc.response();
     rpc_address forward_address;
-    if (!check_status(rpc, &forward_address)) {
+    if (!check_status_and_authz(rpc, &forward_address)) {
         if (!forward_address.is_invalid()) {
             partition_configuration config;
             config.primary = forward_address;
@@ -726,7 +774,7 @@ void meta_service::on_query_configuration_by_index(configuration_query_by_index_
 // meta state thread pool
 void meta_service::on_config_sync(configuration_query_by_node_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -748,7 +796,7 @@ void meta_service::on_config_sync(configuration_query_by_node_rpc rpc)
 void meta_service::on_update_configuration(dsn::message_ex *req)
 {
     configuration_update_response response;
-    if (!check_status_with_msg(req, response)) {
+    if (!check_status_and_authz_with_reply(req, response)) {
         return;
     }
 
@@ -777,7 +825,7 @@ void meta_service::on_update_configuration(dsn::message_ex *req)
 
 void meta_service::on_control_meta_level(configuration_meta_control_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -801,10 +849,7 @@ void meta_service::on_control_meta_level(configuration_meta_control_rpc rpc)
 
 void meta_service::on_propose_balancer(configuration_balancer_rpc rpc)
 {
-    if (!check_status(rpc)) {
-        return;
-    }
-
+    CHECK_APP_ID_STATUS_AND_AUTHZ(rpc.request().gpid.get_app_id());
     const configuration_balancer_request &request = rpc.request();
     LOG_INFO("get proposal balancer request, gpid({})", request.gpid);
     _state->on_propose_balancer(request, rpc.response());
@@ -840,8 +885,8 @@ void meta_service::on_start_recovery(configuration_recovery_rpc rpc)
 
 void meta_service::on_start_restore(dsn::message_ex *req)
 {
-    configuration_create_app_response response;
-    if (!check_status_with_msg(req, response)) {
+    if (!check_status_and_authz_with_reply<configuration_restore_request,
+                                           configuration_create_app_response>(req)) {
         return;
     }
 
@@ -853,7 +898,7 @@ void meta_service::on_start_restore(dsn::message_ex *req)
 void meta_service::on_add_backup_policy(dsn::message_ex *req)
 {
     configuration_add_backup_policy_response response;
-    if (!check_status_with_msg(req, response)) {
+    if (!check_status_and_authz_with_reply(req, response)) {
         return;
     }
 
@@ -871,7 +916,7 @@ void meta_service::on_add_backup_policy(dsn::message_ex *req)
 
 void meta_service::on_query_backup_policy(query_backup_policy_rpc policy_rpc)
 {
-    if (!check_status(policy_rpc)) {
+    if (!check_status_and_authz(policy_rpc)) {
         return;
     }
 
@@ -889,7 +934,7 @@ void meta_service::on_query_backup_policy(query_backup_policy_rpc policy_rpc)
 
 void meta_service::on_modify_backup_policy(configuration_modify_backup_policy_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -906,7 +951,7 @@ void meta_service::on_modify_backup_policy(configuration_modify_backup_policy_rp
 
 void meta_service::on_report_restore_status(configuration_report_restore_status_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -917,10 +962,7 @@ void meta_service::on_report_restore_status(configuration_report_restore_status_
 
 void meta_service::on_query_restore_status(configuration_query_restore_rpc rpc)
 {
-    if (!check_status(rpc)) {
-        return;
-    }
-
+    CHECK_APP_ID_STATUS_AND_AUTHZ(rpc.request().restore_app_id);
     tasking::enqueue(LPC_META_STATE_NORMAL,
                      nullptr,
                      std::bind(&server_state::on_query_restore_status, _state.get(), rpc));
@@ -928,7 +970,7 @@ void meta_service::on_query_restore_status(configuration_query_restore_rpc rpc)
 
 void meta_service::on_add_duplication(duplication_add_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -944,7 +986,7 @@ void meta_service::on_add_duplication(duplication_add_rpc rpc)
 
 void meta_service::on_modify_duplication(duplication_modify_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -960,7 +1002,7 @@ void meta_service::on_modify_duplication(duplication_modify_rpc rpc)
 
 void meta_service::on_query_duplication_info(duplication_query_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -973,7 +1015,7 @@ void meta_service::on_query_duplication_info(duplication_query_rpc rpc)
 
 void meta_service::on_duplication_sync(duplication_sync_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc)) {
         return;
     }
 
@@ -1018,7 +1060,7 @@ void meta_service::initialize_duplication_service()
 
 void meta_service::update_app_env(app_env_rpc env_rpc)
 {
-    if (!check_status(env_rpc)) {
+    if (!check_status_and_authz(env_rpc, nullptr, env_rpc.request().app_name)) {
         return;
     }
 
@@ -1051,10 +1093,7 @@ void meta_service::update_app_env(app_env_rpc env_rpc)
 
 void meta_service::ddd_diagnose(ddd_diagnose_rpc rpc)
 {
-    if (!check_status(rpc)) {
-        return;
-    }
-
+    CHECK_APP_ID_STATUS_AND_AUTHZ(rpc.request().pid.get_app_id());
     auto &response = rpc.response();
     get_partition_guardian()->get_ddd_partitions(rpc.request().pid, response.partitions);
     response.err = ERR_OK;
@@ -1062,7 +1101,7 @@ void meta_service::ddd_diagnose(ddd_diagnose_rpc rpc)
 
 void meta_service::on_start_partition_split(start_split_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
     if (_split_svc == nullptr) {
@@ -1078,7 +1117,7 @@ void meta_service::on_start_partition_split(start_split_rpc rpc)
 
 void meta_service::on_control_partition_split(control_split_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1095,7 +1134,7 @@ void meta_service::on_control_partition_split(control_split_rpc rpc)
 
 void meta_service::on_query_partition_split(query_split_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1109,7 +1148,7 @@ void meta_service::on_query_partition_split(query_split_rpc rpc)
 
 void meta_service::on_register_child_on_meta(register_child_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app.app_name)) {
         return;
     }
 
@@ -1121,7 +1160,7 @@ void meta_service::on_register_child_on_meta(register_child_rpc rpc)
 
 void meta_service::on_notify_stop_split(notify_stop_split_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
     if (_split_svc == nullptr) {
@@ -1137,7 +1176,7 @@ void meta_service::on_notify_stop_split(notify_stop_split_rpc rpc)
 
 void meta_service::on_query_child_state(query_child_state_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
     if (_split_svc == nullptr) {
@@ -1150,7 +1189,7 @@ void meta_service::on_query_child_state(query_child_state_rpc rpc)
 
 void meta_service::on_start_bulk_load(start_bulk_load_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1164,7 +1203,7 @@ void meta_service::on_start_bulk_load(start_bulk_load_rpc rpc)
 
 void meta_service::on_control_bulk_load(control_bulk_load_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1181,7 +1220,7 @@ void meta_service::on_control_bulk_load(control_bulk_load_rpc rpc)
 
 void meta_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1195,7 +1234,7 @@ void meta_service::on_query_bulk_load_status(query_bulk_load_rpc rpc)
 
 void meta_service::on_clear_bulk_load(clear_bulk_load_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1212,9 +1251,7 @@ void meta_service::on_clear_bulk_load(clear_bulk_load_rpc rpc)
 
 void meta_service::on_start_backup_app(start_backup_app_rpc rpc)
 {
-    if (!check_status(rpc)) {
-        return;
-    }
+    CHECK_APP_ID_STATUS_AND_AUTHZ(rpc.request().app_id);
     if (_backup_handler == nullptr) {
         LOG_ERROR("meta doesn't enable backup service");
         rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
@@ -1225,9 +1262,7 @@ void meta_service::on_start_backup_app(start_backup_app_rpc rpc)
 
 void meta_service::on_query_backup_status(query_backup_status_rpc rpc)
 {
-    if (!check_status(rpc)) {
-        return;
-    }
+    CHECK_APP_ID_STATUS_AND_AUTHZ(rpc.request().app_id);
     if (_backup_handler == nullptr) {
         LOG_ERROR("meta doesn't enable backup service");
         rpc.response().err = ERR_SERVICE_NOT_ACTIVE;
@@ -1244,7 +1279,7 @@ size_t meta_service::get_alive_node_count() const
 
 void meta_service::on_start_manual_compact(start_manual_compact_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
     tasking::enqueue(LPC_META_STATE_NORMAL,
@@ -1254,7 +1289,7 @@ void meta_service::on_start_manual_compact(start_manual_compact_rpc rpc)
 
 void meta_service::on_query_manual_compact_status(query_manual_compact_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
     tasking::enqueue(LPC_META_STATE_NORMAL,
@@ -1265,7 +1300,7 @@ void meta_service::on_query_manual_compact_status(query_manual_compact_rpc rpc)
 // ThreadPool: THREAD_POOL_META_SERVER
 void meta_service::on_get_max_replica_count(configuration_get_max_replica_count_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
@@ -1278,7 +1313,7 @@ void meta_service::on_get_max_replica_count(configuration_get_max_replica_count_
 // ThreadPool: THREAD_POOL_META_SERVER
 void meta_service::on_set_max_replica_count(configuration_set_max_replica_count_rpc rpc)
 {
-    if (!check_status(rpc)) {
+    if (!check_status_and_authz(rpc, nullptr, rpc.request().app_name)) {
         return;
     }
 
diff --git a/src/meta/meta_service.h b/src/meta/meta_service.h
index 505b1ca815..58f5a902a1 100644
--- a/src/meta/meta_service.h
+++ b/src/meta/meta_service.h
@@ -57,6 +57,7 @@
 #include "meta_server_failure_detector.h"
 #include "perf_counter/perf_counter_wrapper.h"
 #include "runtime/api_layer1.h"
+#include "runtime/rpc/network.h"
 #include "runtime/rpc/rpc_address.h"
 #include "runtime/rpc/rpc_message.h"
 #include "runtime/rpc/serialization.h"
@@ -65,6 +66,7 @@
 #include "runtime/task/task.h"
 #include "runtime/task/task_code.h"
 #include "runtime/task/task_tracker.h"
+#include "utils/autoref_ptr.h"
 #include "utils/enum_helper.h"
 #include "utils/error_code.h"
 #include "utils/fmt_logging.h"
@@ -73,12 +75,15 @@
 
 namespace dsn {
 class command_deregister;
+
+namespace ranger {
+class ranger_resource_policy_manager;
+} // namespace ranger
 namespace dist {
 class meta_state_service;
 } // namespace dist
 
 namespace replication {
-
 class backup_service;
 class bulk_load_service;
 class meta_duplication_service;
@@ -86,6 +91,7 @@ class meta_split_service;
 class partition_guardian;
 class server_load_balancer;
 class server_state;
+
 namespace mss {
 struct meta_storage;
 } // namespace mss
@@ -143,6 +149,7 @@ class meta_service : public serverlet<meta_service>
     mss::meta_storage *get_meta_storage() const { return _meta_storage.get(); }
 
     server_state *get_server_state() { return _state.get(); }
+    security::access_controller *get_access_controller() { return _access_controller.get(); }
     server_load_balancer *get_balancer() { return _balancer.get(); }
     partition_guardian *get_partition_guardian() { return _partition_guardian.get(); }
     dist::block_service::block_service_manager &get_block_service_manager()
@@ -291,13 +298,29 @@ class meta_service : public serverlet<meta_service>
     meta_leader_state check_leader(dsn::message_ex *req, dsn::rpc_address *forward_address);
     template <typename TRpcHolder>
     meta_leader_state check_leader(TRpcHolder rpc, /*out*/ rpc_address *forward_address);
+
+    // app_name: when the Ranger ACL is enabled, some rpc requests need to verify the app_name
     // ret:
-    //    false: check failed
-    //    true:  check succeed
+    //    false: rpc request check failed because check leader failed or ACL authentication failed
+    //    true:  rpc request check and authentication succeed
     template <typename TRpcHolder>
-    bool check_status(TRpcHolder rpc, /*out*/ rpc_address *forward_address = nullptr);
+    bool check_status_and_authz(TRpcHolder rpc,
+                                /*out*/ rpc_address *forward_address = nullptr,
+                                const std::string &app_name = "");
+
+    // app_name: when the Ranger ACL is enabled, some rpc requests need to verify the app_name
+    // ret:
+    //    false: rpc request check failed because check leader failed or ACL authentication failed
+    //    true:  rpc request check and authentication succeed
     template <typename TRespType>
-    bool check_status_with_msg(message_ex *req, TRespType &response_struct);
+    bool check_status_and_authz_with_reply(message_ex *req,
+                                           TRespType &response_struct,
+                                           const std::string &app_name = "");
+    template <typename TReqType, typename TRespType>
+    bool check_status_and_authz_with_reply(message_ex *msg);
+
+    template <typename TRpcHolder>
+    bool check_leader_status(TRpcHolder rpc, rpc_address *forward_address = nullptr);
 
     error_code remote_storage_initialize();
     bool check_freeze() const;
@@ -319,6 +342,7 @@ class meta_service : public serverlet<meta_service>
     friend class policy_context_test;
     friend class server_state_restore_test;
     friend class test::test_checker;
+    friend class fake_receiver_meta_service;
 
     replication_options _opts;
     meta_options _meta_opts;
@@ -365,7 +389,10 @@ class meta_service : public serverlet<meta_service>
 
     dsn::task_tracker _tracker;
 
-    std::unique_ptr<security::access_controller> _access_controller;
+    std::shared_ptr<security::access_controller> _access_controller;
+
+    // Use Apache Ranger for access control, which is nullptr when not use
+    std::shared_ptr<ranger::ranger_resource_policy_manager> _ranger_resource_policy_manager;
 
     // indicate which operation is processeding in meta server
     std::atomic<meta_op_status> _meta_op_status;
@@ -396,14 +423,8 @@ meta_leader_state meta_service::check_leader(TRpcHolder rpc, rpc_address *forwar
 }
 
 template <typename TRpcHolder>
-bool meta_service::check_status(TRpcHolder rpc, rpc_address *forward_address)
+bool meta_service::check_leader_status(TRpcHolder rpc, rpc_address *forward_address)
 {
-    if (!_access_controller->allowed(rpc.dsn_request())) {
-        rpc.response().err = ERR_ACL_DENY;
-        LOG_INFO("reject request with ERR_ACL_DENY");
-        return false;
-    }
-
     auto result = check_leader(rpc, forward_address);
     if (result == meta_leader_state::kNotLeaderAndCanForwardRpc)
         return false;
@@ -418,20 +439,36 @@ bool meta_service::check_status(TRpcHolder rpc, rpc_address *forward_address)
         LOG_INFO("reject request with {}", rpc.response().err);
         return false;
     }
-
     return true;
 }
 
-template <typename TRespType>
-bool meta_service::check_status_with_msg(message_ex *req, TRespType &response_struct)
+// when the Ranger ACL is enabled, only the leader meta_server will pull Ranger policy, so if it is
+// not the leader, _access_controller may be a null pointer, or a new leader is elected, and the
+// above policy information may be out of date.
+template <typename TRpcHolder>
+bool meta_service::check_status_and_authz(TRpcHolder rpc,
+                                          rpc_address *forward_address,
+                                          const std::string &app_name)
 {
-    if (!_access_controller->allowed(req)) {
-        LOG_INFO("reject request with ERR_ACL_DENY");
-        response_struct.err = ERR_ACL_DENY;
-        reply(req, response_struct);
+    if (!check_leader_status(rpc, forward_address)) {
+        return false;
+    }
+    if (!_access_controller->allowed(rpc.dsn_request(), app_name)) {
+        rpc.response().err = ERR_ACL_DENY;
+        LOG_DEBUG("not authorized {} to operate on app({}) for user({})",
+                  rpc.dsn_request()->rpc_code(),
+                  app_name,
+                  rpc.dsn_request()->io_session->get_client_username());
         return false;
     }
+    return true;
+}
 
+template <typename TRespType>
+bool meta_service::check_status_and_authz_with_reply(message_ex *req,
+                                                     TRespType &response_struct,
+                                                     const std::string &app_name)
+{
     auto result = check_leader(req, nullptr);
     if (result == meta_leader_state::kNotLeaderAndCanForwardRpc) {
         return false;
@@ -444,13 +481,31 @@ bool meta_service::check_status_with_msg(message_ex *req, TRespType &response_st
         } else {
             response_struct.err = ERR_SERVICE_NOT_ACTIVE;
         }
-        LOG_INFO("reject request with {}", response_struct.err);
+        LOG_DEBUG("reject request with {}", response_struct.err);
+        reply(req, response_struct);
+        return false;
+    }
+    if (!_access_controller->allowed(req, app_name)) {
+        response_struct.err = ERR_ACL_DENY;
+        LOG_DEBUG("not authorized {} to operate on app({}) for user({})",
+                  req->rpc_code(),
+                  app_name,
+                  req->io_session->get_client_username());
         reply(req, response_struct);
         return false;
     }
-
     return true;
 }
 
+template <typename TReqType, typename TRespType>
+bool meta_service::check_status_and_authz_with_reply(message_ex *msg)
+{
+    TReqType req;
+    TRespType resp;
+    dsn::message_ex *copied_msg = message_ex::copy_message_no_reply(*msg);
+    dsn::unmarshall(copied_msg, req);
+    return check_status_and_authz_with_reply(msg, resp, req.app_name);
+}
+
 } // namespace replication
 } // namespace dsn
diff --git a/src/meta/server_state.cpp b/src/meta/server_state.cpp
index cb208a6fb1..ecf66a92bc 100644
--- a/src/meta/server_state.cpp
+++ b/src/meta/server_state.cpp
@@ -71,6 +71,7 @@
 #include "runtime/rpc/rpc_address.h"
 #include "runtime/rpc/rpc_message.h"
 #include "runtime/rpc/serialization.h"
+#include "runtime/security/access_controller.h"
 #include "runtime/task/async_calls.h"
 #include "runtime/task/task.h"
 #include "runtime/task/task_spec.h"
@@ -1438,14 +1439,17 @@ void server_state::recall_app(dsn::message_ex *msg)
 }
 
 void server_state::list_apps(const configuration_list_apps_request &request,
-                             configuration_list_apps_response &response)
+                             configuration_list_apps_response &response,
+                             dsn::message_ex *msg) const
 {
     LOG_DEBUG("list app request, status({})", request.status);
     zauto_read_lock l(_lock);
-    for (auto &kv : _all_apps) {
+    for (const auto &kv : _all_apps) {
         app_state &app = *(kv.second);
         if (request.status == app_status::AS_INVALID || request.status == app.status) {
-            response.infos.push_back(app);
+            if (nullptr == msg || _meta_svc->get_access_controller()->allowed(msg, app.app_name)) {
+                response.infos.push_back(app);
+            }
         }
     }
     response.err = dsn::ERR_OK;
diff --git a/src/meta/server_state.h b/src/meta/server_state.h
index 944118c695..a3feb28f9a 100644
--- a/src/meta/server_state.h
+++ b/src/meta/server_state.h
@@ -173,7 +173,8 @@ class server_state
     void recall_app(dsn::message_ex *msg);
     void rename_app(configuration_rename_app_rpc rpc);
     void list_apps(const configuration_list_apps_request &request,
-                   configuration_list_apps_response &response);
+                   configuration_list_apps_response &response,
+                   dsn::message_ex *msg = nullptr) const;
     void restore_app(dsn::message_ex *msg);
 
     // app env operations
diff --git a/src/meta/server_state_restore.cpp b/src/meta/server_state_restore.cpp
index 43cfb79850..93d790e66b 100644
--- a/src/meta/server_state_restore.cpp
+++ b/src/meta/server_state_restore.cpp
@@ -239,30 +239,25 @@ void server_state::on_query_restore_status(configuration_query_restore_rpc rpc)
     response.err = ERR_OK;
 
     std::shared_ptr<app_state> app = get_app(request.restore_app_id);
-    if (app == nullptr) {
-        response.err = ERR_APP_NOT_EXIST;
-    } else {
-        if (app->status == app_status::AS_DROPPED) {
-            response.err = ERR_APP_DROPPED;
-        } else {
-            response.restore_progress.resize(app->partition_count,
-                                             cold_backup_constant::PROGRESS_FINISHED);
-            response.restore_status.resize(app->partition_count, ERR_OK);
-            for (int32_t i = 0; i < app->partition_count; i++) {
-                const auto &r_state = app->helpers->restore_states[i];
-                const auto &p = app->partitions[i];
-                if (!p.primary.is_invalid() || !p.secondaries.empty()) {
-                    // already have primary, restore succeed
-                    continue;
-                } else {
-                    if (r_state.progress < response.restore_progress[i]) {
-                        response.restore_progress[i] = r_state.progress;
-                    }
-                }
-                response.restore_status[i] = r_state.restore_status;
-            }
+    CHECK(app, "app must be valid");
+    if (app->status == app_status::AS_DROPPED) {
+        response.err = ERR_APP_DROPPED;
+        return;
+    }
+    response.restore_progress.resize(app->partition_count, cold_backup_constant::PROGRESS_FINISHED);
+    response.restore_status.resize(app->partition_count, ERR_OK);
+    for (int32_t i = 0; i < app->partition_count; i++) {
+        const auto &r_state = app->helpers->restore_states[i];
+        const auto &p = app->partitions[i];
+        if (!p.primary.is_invalid() || !p.secondaries.empty()) {
+            // already have primary, restore succeed
+            continue;
+        }
+        if (r_state.progress < response.restore_progress[i]) {
+            response.restore_progress[i] = r_state.progress;
         }
+        response.restore_status[i] = r_state.restore_status;
     }
 }
-}
-}
+} // namespace replication
+} // namespace dsn
diff --git a/src/meta/test/meta_service_test.cpp b/src/meta/test/meta_service_test.cpp
index 96b9fde914..bf993c0dde 100644
--- a/src/meta/test/meta_service_test.cpp
+++ b/src/meta/test/meta_service_test.cpp
@@ -55,8 +55,7 @@ class meta_service_test : public meta_test_base
             rpc_address leader;
             auto rpc = create_fake_rpc();
             rpc.dsn_request()->header->context.u.is_forward_supported = false;
-            bool res = _ms->check_status(rpc, &leader);
-            ASSERT_EQ(false, res);
+            ASSERT_FALSE(_ms->check_status_and_authz(rpc, &leader));
             ASSERT_EQ(ERR_FORWARD_TO_OTHERS, rpc.response().err);
             ASSERT_EQ(leader.to_std_string(), "1.2.3.4:10086");
             ASSERT_EQ(app_env_rpc::forward_mail_box().size(), 0);
@@ -66,8 +65,7 @@ class meta_service_test : public meta_test_base
         RPC_MOCKING(app_env_rpc)
         {
             auto rpc = create_fake_rpc();
-            bool res = _ms->check_status(rpc);
-            ASSERT_EQ(false, res);
+            ASSERT_FALSE(_ms->check_status_and_authz(rpc));
             ASSERT_EQ(app_env_rpc::forward_mail_box().size(), 1);
             ASSERT_EQ(app_env_rpc::forward_mail_box()[0].remote_address().to_std_string(),
                       "1.2.3.4:10086");
@@ -85,8 +83,7 @@ class meta_service_test : public meta_test_base
         {
             rpc_address leader;
             auto rpc = create_fake_rpc();
-            auto res = _ms->check_status(rpc, &leader);
-            ASSERT_EQ(true, res);
+            ASSERT_TRUE(_ms->check_status_and_authz(rpc, &leader));
             ASSERT_EQ(app_env_rpc::forward_mail_box().size(), 0);
         }
 
diff --git a/src/meta/test/meta_service_test_app.h b/src/meta/test/meta_service_test_app.h
index a931854ad8..b755755919 100644
--- a/src/meta/test/meta_service_test_app.h
+++ b/src/meta/test/meta_service_test_app.h
@@ -88,7 +88,10 @@ inline dsn::message_ex *create_corresponding_receive(dsn::message_ex *request_ms
 class fake_receiver_meta_service : public dsn::replication::meta_service
 {
 public:
-    fake_receiver_meta_service() : meta_service() {}
+    fake_receiver_meta_service() : meta_service()
+    {
+        _access_controller = security::create_meta_access_controller(nullptr);
+    }
     virtual ~fake_receiver_meta_service() {}
     virtual void reply_message(dsn::message_ex *request, dsn::message_ex *response) override
     {
diff --git a/src/replica/CMakeLists.txt b/src/replica/CMakeLists.txt
index fa4365235b..7e27ef9b77 100644
--- a/src/replica/CMakeLists.txt
+++ b/src/replica/CMakeLists.txt
@@ -69,6 +69,7 @@ set(MY_PROJ_LIBS
     dsn_http
     dsn_runtime
     dsn_aio
+    dsn_meta_server
     galaxy-fds-sdk-cpp
     PocoNet
     PocoFoundation
diff --git a/src/replica/backup/test/CMakeLists.txt b/src/replica/backup/test/CMakeLists.txt
index 49806f0142..21ec10f505 100644
--- a/src/replica/backup/test/CMakeLists.txt
+++ b/src/replica/backup/test/CMakeLists.txt
@@ -32,7 +32,7 @@ set(MY_PROJ_LIBS dsn_meta_server
         gtest
 )
 
-set(MY_BOOST_LIBS Boost::system Boost::filesystem)
+set(MY_BOOST_LIBS Boost::system Boost::filesystem Boost::regex)
 
 set(MY_BINPLACES
         config-test.ini
diff --git a/src/runtime/ranger/ranger_resource_policy_manager.cpp b/src/runtime/ranger/ranger_resource_policy_manager.cpp
index 3c30e262ee..440bcee718 100644
--- a/src/runtime/ranger/ranger_resource_policy_manager.cpp
+++ b/src/runtime/ranger/ranger_resource_policy_manager.cpp
@@ -46,6 +46,7 @@
 #include "ranger_resource_policy_manager.h"
 #include "rapidjson/allocators.h"
 #include "runtime/ranger/ranger_resource_policy.h"
+#include "runtime/ranger/ranger_resource_policy_manager.h"
 #include "runtime/task/async_calls.h"
 #include "runtime/task/task.h"
 #include "runtime/task/task_code.h"
@@ -123,6 +124,20 @@ const std::map<std::string, access_type> kAccessTypeMaping({{"READ", access_type
                                                             {"LIST", access_type::kList},
                                                             {"METADATA", access_type::kMetadata},
                                                             {"CONTROL", access_type::kControl}});
+
+// Pull policies in JSON format from Ranger service.
+dsn::error_code pull_policies_from_ranger_service(std::string *ranger_policies)
+{
+    std::string cmd =
+        fmt::format("curl {}/{}", FLAGS_ranger_service_url, FLAGS_ranger_service_name);
+    std::stringstream resp;
+    if (dsn::utils::pipe_execute(cmd.c_str(), resp) != 0) {
+        return dsn::ERR_SYNC_RANGER_POLICIES_FAILED;
+    }
+
+    *ranger_policies = resp.str();
+    return dsn::ERR_OK;
+}
 } // anonymous namespace
 
 const std::chrono::milliseconds kLoadRangerPolicyRetryDelayMs(10000);
@@ -131,15 +146,12 @@ ranger_resource_policy_manager::ranger_resource_policy_manager(
     dsn::replication::meta_service *meta_svc)
     : _meta_svc(meta_svc), _local_policy_version(-1)
 {
-    _ranger_policy_meta_root = dsn::replication::meta_options::concat_path_unix_style(
-        _meta_svc->cluster_root(), "ranger_policy_meta_root");
-
-    // GLOBAL - KMetadata
+    // GLOBAL - kMetadata
     register_rpc_access_type(
         access_type::kMetadata,
         {"RPC_CM_LIST_NODES", "RPC_CM_CLUSTER_INFO", "RPC_CM_LIST_APPS", "RPC_QUERY_DISK_INFO"},
         _ac_type_of_global_rpcs);
-    // GLOBAL - KControl
+    // GLOBAL - kControl
     register_rpc_access_type(access_type::kControl,
                              {"RPC_HTTP_SERVICE",
                               "RPC_CM_CONTROL_META",
@@ -149,15 +161,15 @@ ranger_resource_policy_manager::ranger_resource_policy_manager(
                               "RPC_DETECT_HOTKEY",
                               "RPC_CLI_CLI_CALL_ACK"},
                              _ac_type_of_global_rpcs);
-    // DATABASE - KList
+    // DATABASE - kList
     register_rpc_access_type(access_type::kList, {"RPC_CM_LIST_APPS"}, _ac_type_of_database_rpcs);
-    // DATABASE - KCreate
+    // DATABASE - kCreate
     register_rpc_access_type(
         access_type::kCreate, {"RPC_CM_CREATE_APP"}, _ac_type_of_database_rpcs);
-    // DATABASE - KDrop
+    // DATABASE - kDrop
     register_rpc_access_type(
         access_type::kDrop, {"RPC_CM_DROP_APP", "RPC_CM_RECALL_APP"}, _ac_type_of_database_rpcs);
-    // DATABASE - KMetadata
+    // DATABASE - kMetadata
     register_rpc_access_type(access_type::kMetadata,
                              {"RPC_CM_QUERY_BACKUP_STATUS",
                               "RPC_CM_QUERY_RESTORE_STATUS",
@@ -167,7 +179,7 @@ ranger_resource_policy_manager::ranger_resource_policy_manager(
                               "RPC_CM_QUERY_MANUAL_COMPACT_STATUS",
                               "RPC_CM_GET_MAX_REPLICA_COUNT"},
                              _ac_type_of_database_rpcs);
-    // DATABASE - KControl
+    // DATABASE - kControl
     register_rpc_access_type(access_type::kControl,
                              {"RPC_CM_START_BACKUP_APP",
                               "RPC_CM_START_RESTORE",
@@ -187,6 +199,73 @@ ranger_resource_policy_manager::ranger_resource_policy_manager(
                              _ac_type_of_database_rpcs);
 }
 
+void ranger_resource_policy_manager::start()
+{
+    CHECK_NOTNULL(_meta_svc, "");
+    _ranger_policy_meta_root = dsn::replication::meta_options::concat_path_unix_style(
+        _meta_svc->cluster_root(), "ranger_policy_meta_root");
+    tasking::enqueue_timer(LPC_USE_RANGER_ACCESS_CONTROL,
+                           &_tracker,
+                           [this]() { this->update_policies_from_ranger_service(); },
+                           std::chrono::seconds(FLAGS_update_ranger_policy_interval_sec),
+                           0,
+                           std::chrono::milliseconds(1));
+}
+
+bool ranger_resource_policy_manager::allowed(const int rpc_code,
+                                             const std::string &user_name,
+                                             const std::string &database_name)
+{
+    do {
+        const auto &ac_type = _ac_type_of_global_rpcs.find(rpc_code);
+        // It's not a GLOBAL rpc code.
+        if (ac_type == _ac_type_of_global_rpcs.end()) {
+            break;
+        }
+
+        // Check if it is allowed by any GLOBAL policy.
+        utils::auto_read_lock l(_global_policies_lock);
+        for (const auto &policy : _global_policies_cache) {
+            if (policy.policies.allowed(ac_type->second, user_name)) {
+                return true;
+            }
+        }
+
+        // It's not allowed to access except list_app.
+        // list_app rpc code is in both GLOBAL and DATABASE policies, check the DATABASE policies
+        // later.
+        if (rpc_code != RPC_CM_LIST_APPS.code()) {
+            return false;
+        }
+    } while (false);
+
+    do {
+        const auto &ac_type = _ac_type_of_database_rpcs.find(rpc_code);
+        // It's not a DATABASE rpc code.
+        if (ac_type == _ac_type_of_database_rpcs.end()) {
+            break;
+        }
+
+        // Check if it is allowed by any DATABASE policy.
+        utils::auto_read_lock l(_database_policies_lock);
+        for (const auto &policy : _database_policies_cache) {
+            if (!policy.policies.allowed(ac_type->second, user_name)) {
+                continue;
+            }
+            // Legacy tables may don't contain database section.
+            if (database_name.empty() && policy.database_names.count("*") != 0) {
+                return true;
+            }
+            if (policy.database_names.count(database_name) != 0) {
+                return true;
+            }
+        }
+    } while (false);
+
+    // The check that does not match any policy returns false.
+    return false;
+}
+
 void ranger_resource_policy_manager::parse_policies_from_json(const rapidjson::Value &data,
                                                               std::vector<policy_item> &policies)
 {
@@ -239,20 +318,6 @@ dsn::error_code ranger_resource_policy_manager::update_policies_from_ranger_serv
     return dsn::ERR_OK;
 }
 
-dsn::error_code ranger_resource_policy_manager::pull_policies_from_ranger_service(
-    std::string *ranger_policies) const
-{
-    std::string cmd =
-        fmt::format("curl {}/{}", FLAGS_ranger_service_url, FLAGS_ranger_service_name);
-    std::stringstream resp;
-    if (dsn::utils::pipe_execute(cmd.c_str(), resp) != 0) {
-        return dsn::ERR_SYNC_RANGER_POLICIES_FAILED;
-    }
-
-    *ranger_policies = resp.str();
-    return dsn::ERR_OK;
-}
-
 dsn::error_code ranger_resource_policy_manager::load_policies_from_json(const std::string &data)
 {
     // The Ranger policy pulled from Ranger service demo.
diff --git a/src/runtime/ranger/ranger_resource_policy_manager.h b/src/runtime/ranger/ranger_resource_policy_manager.h
index 5c920d9121..3b548bc67b 100644
--- a/src/runtime/ranger/ranger_resource_policy_manager.h
+++ b/src/runtime/ranger/ranger_resource_policy_manager.h
@@ -71,6 +71,12 @@ class ranger_resource_policy_manager
 
     ~ranger_resource_policy_manager() = default;
 
+    // When using Ranger for ACL, periodically pull policies from Ranger service.
+    void start();
+
+    // Return true if the 'user_name' is allowed to access 'app_name' via 'rpc_code'.
+    bool allowed(const int rpc_code, const std::string &user_name, const std::string &app_name);
+
 private:
     // Parse Ranger ACL policies from 'data' in JSON format into 'policies'.
     static void parse_policies_from_json(const rapidjson::Value &data,
@@ -79,9 +85,6 @@ class ranger_resource_policy_manager
     // Update policies from Ranger service.
     dsn::error_code update_policies_from_ranger_service();
 
-    // Pull policies in JSON format from Ranger service.
-    dsn::error_code pull_policies_from_ranger_service(std::string *ranger_policies) const;
-
     // Load policies from JSON formated string.
     dsn::error_code load_policies_from_json(const std::string &data);
 
@@ -100,6 +103,13 @@ class ranger_resource_policy_manager
     // Sync policies to app_envs(REPLICA_ACCESS_CONTROLLER_RANGER_POLICIES).
     dsn::error_code sync_policies_to_app_envs();
 
+protected:
+    // The cache of the global resources policies, it's a subset of '_all_resource_policies'.
+    resource_policies _global_policies_cache;
+
+    // The cache of the database resources policies, it's a subset of '_all_resource_policies'.
+    resource_policies _database_policies_cache;
+
 private:
     dsn::task_tracker _tracker;
 
@@ -107,16 +117,8 @@ class ranger_resource_policy_manager
     std::string _ranger_policy_meta_root;
 
     replication::meta_service *_meta_svc;
-
-    // The cache of the global resources policies, it's a subset of '_all_resource_policies'.
-    utils::rw_lock_nr _global_policies_lock; // [
-    resource_policies _global_policies_cache;
-    // ]
-
-    // The cache of the database resources policies, it's a subset of '_all_resource_policies'.
-    utils::rw_lock_nr _database_policies_lock; // [
-    resource_policies _database_policies_cache;
-    // ]
+    utils::rw_lock_nr _global_policies_lock;
+    utils::rw_lock_nr _database_policies_lock;
 
     // The access type of RPCs which access global level resources.
     access_type_of_rpc_code _ac_type_of_global_rpcs;
diff --git a/src/runtime/security/access_controller.cpp b/src/runtime/security/access_controller.cpp
index 4516ed0f30..56e9acd7e6 100644
--- a/src/runtime/security/access_controller.cpp
+++ b/src/runtime/security/access_controller.cpp
@@ -61,9 +61,10 @@ bool access_controller::is_super_user(const std::string &user_name) const
     return _super_users.find(user_name) != _super_users.end();
 }
 
-std::unique_ptr<access_controller> create_meta_access_controller()
+std::shared_ptr<access_controller> create_meta_access_controller(
+    const std::shared_ptr<ranger::ranger_resource_policy_manager> &policy_manager)
 {
-    return std::make_unique<meta_access_controller>();
+    return std::make_shared<meta_access_controller>(policy_manager);
 }
 
 std::unique_ptr<access_controller> create_replica_access_controller(const std::string &name)
diff --git a/src/runtime/security/access_controller.h b/src/runtime/security/access_controller.h
index 81d3e91da9..5ae6224946 100644
--- a/src/runtime/security/access_controller.h
+++ b/src/runtime/security/access_controller.h
@@ -26,6 +26,10 @@
 namespace dsn {
 class message_ex;
 
+namespace ranger {
+class ranger_resource_policy_manager;
+}
+
 namespace security {
 
 class access_controller
@@ -66,7 +70,8 @@ class access_controller
     std::unordered_set<std::string> _super_users;
 };
 
-std::unique_ptr<access_controller> create_meta_access_controller();
+std::shared_ptr<access_controller> create_meta_access_controller(
+    const std::shared_ptr<ranger::ranger_resource_policy_manager> &policy_manager);
 
 std::unique_ptr<access_controller>
 create_replica_access_controller(const std::string &replica_name);
diff --git a/src/runtime/security/meta_access_controller.cpp b/src/runtime/security/meta_access_controller.cpp
index 2933ce0ece..8adf81f755 100644
--- a/src/runtime/security/meta_access_controller.cpp
+++ b/src/runtime/security/meta_access_controller.cpp
@@ -19,6 +19,7 @@
 
 #include <vector>
 
+#include "runtime/ranger/ranger_resource_policy_manager.h"
 #include "runtime/rpc/network.h"
 #include "runtime/rpc/rpc_message.h"
 #include "runtime/task/task_code.h"
@@ -34,21 +35,59 @@ DSN_DEFINE_string(security,
                   "",
                   "allowed list of rpc codes for meta_access_controller");
 
-meta_access_controller::meta_access_controller()
+DSN_DECLARE_bool(enable_acl);
+DSN_DECLARE_bool(enable_ranger_acl);
+
+meta_access_controller::meta_access_controller(
+    const std::shared_ptr<ranger::ranger_resource_policy_manager> &policy_manager)
+    : _ranger_resource_policy_manager(policy_manager)
 {
     // MetaServer serves the allow-list RPC from all users. RPCs unincluded are accessible to only
     // superusers.
     if (utils::is_empty(FLAGS_meta_acl_rpc_allow_list)) {
-        register_allowed_list("RPC_CM_LIST_APPS");
-        register_allowed_list("RPC_CM_LIST_NODES");
-        register_allowed_list("RPC_CM_CLUSTER_INFO");
-        register_allowed_list("RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX");
+        register_allowed_rpc_code_list({"RPC_CM_LIST_APPS",
+                                        "RPC_CM_LIST_NODES",
+                                        "RPC_CM_CLUSTER_INFO",
+                                        "RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX"});
     } else {
         std::vector<std::string> rpc_code_white_list;
         utils::split_args(FLAGS_meta_acl_rpc_allow_list, rpc_code_white_list, ',');
-        for (const auto &rpc_code : rpc_code_white_list) {
-            register_allowed_list(rpc_code);
-        }
+        register_allowed_rpc_code_list(rpc_code_white_list);
+    }
+
+    // use Ranger policy
+    if (FLAGS_enable_ranger_acl) {
+        register_allowed_rpc_code_list({"RPC_CM_UPDATE_PARTITION_CONFIGURATION",
+                                        "RPC_CM_CONFIG_SYNC",
+                                        "RPC_CM_DUPLICATION_SYNC",
+                                        "RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX",
+                                        "RPC_CM_REPORT_RESTORE_STATUS",
+                                        "RPC_CM_NOTIFY_STOP_SPLIT",
+                                        "RPC_CM_QUERY_CHILD_STATE",
+                                        "RPC_NEGOTIATION",
+                                        "RPC_CALL_RAW_MESSAGE",
+                                        "RPC_CALL_RAW_SESSION_DISCONNECT",
+                                        "RPC_NFS_GET_FILE_SIZE",
+                                        "RPC_FD_FAILURE_DETECTOR_PING",
+                                        "RPC_CALL_RAW_MESSAGE",
+                                        "RPC_CALL_RAW_SESSION_DISCONNECT",
+                                        "RPC_CONFIG_PROPOSAL",
+                                        "RPC_GROUP_CHECK",
+                                        "RPC_QUERY_REPLICA_INFO",
+                                        "RPC_QUERY_LAST_CHECKPOINT_INFO",
+                                        "RPC_PREPARE",
+                                        "RPC_GROUP_CHECK",
+                                        "RPC_QUERY_APP_INFO",
+                                        "RPC_LEARN_COMPLETION_NOTIFY",
+                                        "RPC_LEARN_ADD_LEARNER",
+                                        "RPC_REMOVE_REPLICA",
+                                        "RPC_COLD_BACKUP",
+                                        "RPC_CLEAR_COLD_BACKUP",
+                                        "RPC_SPLIT_NOTIFY_CATCH_UP",
+                                        "RPC_SPLIT_UPDATE_CHILD_PARTITION_COUNT",
+                                        "RPC_BULK_LOAD",
+                                        "RPC_GROUP_BULK_LOAD"});
+        _ranger_resource_policy_manager->start();
     }
 }
 
@@ -61,15 +100,45 @@ bool meta_access_controller::allowed(message_ex *msg)
     return false;
 }
 
-void meta_access_controller::register_allowed_list(const std::string &rpc_code)
+bool meta_access_controller::allowed(message_ex *msg, const std::string &app_name)
 {
-    auto code = task_code::try_get(rpc_code, TASK_CODE_INVALID);
-    CHECK_NE_MSG(code,
-                 TASK_CODE_INVALID,
-                 "invalid task code({}) in rpc_code_white_list of security section",
-                 rpc_code);
+    const auto rpc_code = msg->rpc_code().code();
+    const auto &user_name = msg->io_session->get_client_username();
+
+    // when the Ranger ACL is not enabled, the old ACL will be used in these three cases, the ACL
+    // will be allowed:
+    // 1. enable_acl is false
+    // 2. the user_name is super user
+    // 3. the rpc_code is in _allowed_rpc_code_list
+    if (!FLAGS_enable_ranger_acl) {
+        return !FLAGS_enable_acl || is_super_user(user_name) ||
+               _allowed_rpc_code_list.find(rpc_code) != _allowed_rpc_code_list.end();
+    }
+
+    // in this case, the Ranger ACL is enabled. In both cases, the ACL will be allowed:
+    // 1. the rpc_code is in _allowed_rpc_code_list.(usually internal rpc)
+    // 2. the user_name and resource have passed the validation of Ranger policy
+    if (_allowed_rpc_code_list.find(rpc_code) != _allowed_rpc_code_list.end()) {
+        return true;
+    }
+    std::string database_name = ranger::get_database_name_from_app_name(app_name);
+    LOG_DEBUG("Ranger access controller with user_name = {}, rpc = {}, database_name = {}",
+              user_name,
+              msg->rpc_code(),
+              database_name);
+    return _ranger_resource_policy_manager->allowed(rpc_code, user_name, database_name);
+}
 
-    _allowed_rpc_code_list.insert(code);
+void meta_access_controller::register_allowed_rpc_code_list(
+    const std::vector<std::string> &rpc_list)
+{
+    _allowed_rpc_code_list.clear();
+    for (const auto &rpc_code : rpc_list) {
+        auto code = task_code::try_get(rpc_code, TASK_CODE_INVALID);
+        CHECK_NE_MSG(code, TASK_CODE_INVALID, "invalid task code.");
+        _allowed_rpc_code_list.insert(code);
+    }
 }
+
 } // namespace security
 } // namespace dsn
diff --git a/src/runtime/security/meta_access_controller.h b/src/runtime/security/meta_access_controller.h
index 5f24231a6b..0def5bc497 100644
--- a/src/runtime/security/meta_access_controller.h
+++ b/src/runtime/security/meta_access_controller.h
@@ -17,26 +17,38 @@
 
 #pragma once
 
+#include <memory>
 #include <string>
 #include <unordered_set>
+#include <vector>
 
 #include "access_controller.h"
 
 namespace dsn {
 class message_ex;
 
+namespace ranger {
+class ranger_resource_policy_manager;
+}
+
 namespace security {
 
 class meta_access_controller : public access_controller
 {
 public:
-    meta_access_controller();
+    meta_access_controller(
+        const std::shared_ptr<ranger::ranger_resource_policy_manager> &policy_manager);
+
     bool allowed(message_ex *msg) override;
 
+    bool allowed(message_ex *msg, const std::string &app_name = "") override;
+
 private:
-    void register_allowed_list(const std::string &rpc_code);
+    void register_allowed_rpc_code_list(const std::vector<std::string> &rpc_list);
 
     std::unordered_set<int> _allowed_rpc_code_list;
+
+    std::shared_ptr<ranger::ranger_resource_policy_manager> _ranger_resource_policy_manager;
 };
 } // namespace security
 } // namespace dsn
diff --git a/src/runtime/test/meta_access_controller_test.cpp b/src/runtime/test/meta_access_controller_test.cpp
index 895348110f..822c5a0511 100644
--- a/src/runtime/test/meta_access_controller_test.cpp
+++ b/src/runtime/test/meta_access_controller_test.cpp
@@ -39,24 +39,27 @@ DSN_DECLARE_bool(enable_acl);
 class meta_access_controller_test : public testing::Test
 {
 public:
-    meta_access_controller_test() { _meta_access_controller = create_meta_access_controller(); }
+    meta_access_controller_test()
+    {
+        _meta_access_controller = create_meta_access_controller(nullptr);
+    }
 
     void set_super_user(const std::string &super_user)
     {
         _meta_access_controller->_super_users.insert(super_user);
     }
 
-    bool pre_check(const std::string &user_name)
+    bool is_super_user_or_disable_acl(const std::string &user_name)
     {
-        return _meta_access_controller->pre_check(user_name);
+        return !FLAGS_enable_acl || _meta_access_controller->is_super_user(user_name);
     }
 
     bool allowed(dsn::message_ex *msg) { return _meta_access_controller->allowed(msg); }
 
-    std::unique_ptr<access_controller> _meta_access_controller;
+    std::shared_ptr<access_controller> _meta_access_controller;
 };
 
-TEST_F(meta_access_controller_test, pre_check)
+TEST_F(meta_access_controller_test, is_super_user_or_disable_acl)
 {
     const std::string SUPER_USER_NAME = "super_user";
     struct
@@ -73,7 +76,7 @@ TEST_F(meta_access_controller_test, pre_check)
 
     for (const auto &test : tests) {
         FLAGS_enable_acl = test.enable_acl;
-        ASSERT_EQ(pre_check(test.user_name), test.result);
+        ASSERT_EQ(is_super_user_or_disable_acl(test.user_name), test.result);
     }
 
     FLAGS_enable_acl = origin_enable_acl;
diff --git a/src/runtime/test/ranger_resource_policy_manager_test.cpp b/src/runtime/test/ranger_resource_policy_manager_test.cpp
index 475808e867..8bf4e3f56a 100644
--- a/src/runtime/test/ranger_resource_policy_manager_test.cpp
+++ b/src/runtime/test/ranger_resource_policy_manager_test.cpp
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <fmt/core.h>
 #include <gtest/gtest-message.h>
 #include <gtest/gtest-test-part.h>
 #include <gtest/gtest.h>
@@ -31,6 +32,7 @@
 #include "runtime/ranger/access_type.h"
 #include "runtime/ranger/ranger_resource_policy.h"
 #include "runtime/ranger/ranger_resource_policy_manager.h"
+#include "runtime/task/task_code.h"
 #include "utils/blob.h"
 
 namespace dsn {
@@ -143,18 +145,15 @@ TEST(ranger_resource_policy_manager_test, parse_policies_from_json_for_test)
 TEST(ranger_resource_policy_manager_test, ranger_resource_policy_serialized_test)
 {
     // 1. Create a fake resource policies data in 'fake_all_resource_policies'
-    acl_policies fake_policy;
-    fake_policy.allow_policies = {{access_type::kRead | access_type::kWrite | access_type::kList,
-                                   {"user1", "user2", "user3", "user4"}}};
-    fake_policy.allow_policies_exclude = {{access_type::kWrite | access_type::kCreate, {"user2"}}};
-    fake_policy.deny_policies = {{access_type::kRead | access_type::kWrite, {"user3", "user4"}}};
-    fake_policy.deny_policies_exclude = {{access_type::kRead | access_type::kList, {"user4"}}};
-
-    ranger_resource_policy fake_ranger_resource_policy;
-    fake_ranger_resource_policy.name = "pegasus_ranger_test";
-    fake_ranger_resource_policy.database_names = {"database1", "database2"};
-    fake_ranger_resource_policy.table_names = {"database1_table", "database2_table"};
-    fake_ranger_resource_policy.policies = fake_policy;
+    ranger_resource_policy fake_ranger_resource_policy(
+        {"pegasus_ranger_test",
+         {"database1", "database2"},
+         {"database1_table", "database2_table"},
+         {{{access_type::kRead | access_type::kWrite | access_type::kList,
+            {"user1", "user2", "user3", "user4"}}},
+          {{access_type::kWrite | access_type::kCreate, {"user2"}}},
+          {{access_type::kRead | access_type::kWrite, {"user3", "user4"}}},
+          {{access_type::kRead | access_type::kList, {"user4"}}}}});
 
     std::string resource_type_name = enum_to_string(resource_type::kDatabaseTable);
     all_resource_policies fake_all_resource_policies{
@@ -256,5 +255,108 @@ TEST(ranger_resource_policy_manager_test, get_table_name_from_app_name_test)
     }
 }
 
+class ranger_resource_policy_manager_function_test : public ranger_resource_policy_manager,
+                                                     public testing::Test
+{
+public:
+    ranger_resource_policy_manager_function_test() : ranger_resource_policy_manager(nullptr)
+    {
+        ranger_resource_policy fake_ranger_resource_policy_1(
+            {"",
+             {"database1"},
+             {},
+             {{{access_type::kList | access_type::kMetadata, {"user1", "user2"}}},
+              {{access_type::kMetadata, {"user2"}}},
+              {},
+              {}}});
+        ranger_resource_policy fake_ranger_resource_policy_2(
+            {"",
+             {"database2"},
+             {},
+             {{{access_type::kCreate | access_type::kDrop | access_type::kControl,
+                {"user3", "user4"}}},
+              {{access_type::kControl, {"user4"}}},
+              {},
+              {}}});
+        ranger_resource_policy fake_ranger_resource_policy_3(
+            {"",
+             {"*"},
+             {},
+             {{{access_type::kCreate, {"user5", "user6"}}},
+              {{access_type::kCreate, {"user6"}}},
+              {},
+              {}}});
+        _database_policies_cache = {fake_ranger_resource_policy_1,
+                                    fake_ranger_resource_policy_2,
+                                    fake_ranger_resource_policy_3};
+
+        ranger_resource_policy fake_ranger_resource_policy_4(
+            {"",
+             {"database3"},
+             {},
+             {{{access_type::kMetadata, {"user7", "user8"}}},
+              {{access_type::kMetadata, {"user8"}}},
+              {},
+              {}}});
+        ranger_resource_policy fake_ranger_resource_policy_5(
+            {"",
+             {"database4"},
+             {},
+             {{{access_type::kControl, {"user9", "user10"}}},
+              {{access_type::kControl, {"user10"}}},
+              {},
+              {}}});
+        _global_policies_cache = {fake_ranger_resource_policy_4, fake_ranger_resource_policy_5};
+    }
+};
+
+TEST_F(ranger_resource_policy_manager_function_test, allowed)
+{
+    struct test_case
+    {
+        std::string rpc_code;
+        std::string user_name;
+        std::string database_name;
+        bool expected_result;
+    } tests[] = {{"TASK_CODE_INVALID", "user1", "database1", false},
+                 {"RPC_CM_CREATE_APP", "user1", "database1", false},
+                 {"RPC_CM_CREATE_APP", "user2", "database1", false},
+                 {"RPC_CM_LIST_APPS", "user1", "database1", true},
+                 {"RPC_CM_LIST_APPS", "user2", "database1", true},
+                 {"RPC_CM_GET_MAX_REPLICA_COUNT", "user1", "database1", true},
+                 {"RPC_CM_GET_MAX_REPLICA_COUNT", "user2", "database1", false},
+                 {"TASK_CODE_INVALID", "user3", "database2", false},
+                 {"RPC_CM_CREATE_APP", "user3", "database2", true},
+                 {"RPC_CM_CREATE_APP", "user4", "database2", true},
+                 {"RPC_CM_START_BACKUP_APP", "user3", "database2", true},
+                 {"RPC_CM_START_BACKUP_APP", "user4", "database2", false},
+                 {"TASK_CODE_INVALID", "user5", "", false},
+                 {"RPC_CM_CREATE_APP", "user5", "", true},
+                 {"RPC_CM_CREATE_APP", "user5", "database2", false},
+                 {"RPC_CM_CREATE_APP", "user6", "", false},
+                 {"RPC_CM_CREATE_APP", "user6", "database2", false},
+                 {"TASK_CODE_INVALID", "user7", "database3", false},
+                 {"RPC_CM_LIST_NODES", "user7", "database3", true},
+                 {"RPC_CM_LIST_NODES", "user8", "database3", false},
+                 {"RPC_CM_LIST_APPS", "user7", "database3", true},
+                 {"RPC_CM_LIST_APPS", "user8", "database3", false},
+                 {"TASK_CODE_INVALID", "user9", "database4", false},
+                 {"RPC_CM_LIST_NODES", "user9", "database4", false},
+                 {"RPC_CM_LIST_NODES", "user10", "database4", false},
+                 {"RPC_CM_LIST_APPS", "user9", "database4", false},
+                 {"RPC_CM_LIST_APPS", "user10", "database4", false},
+                 {"RPC_CM_CONTROL_META", "user9", "database4", true},
+                 {"RPC_CM_CONTROL_META", "user10", "database4", false}};
+    for (const auto &test : tests) {
+        auto code = task_code::try_get(test.rpc_code, TASK_CODE_INVALID);
+        auto actual_result = allowed(code, test.user_name, test.database_name);
+        EXPECT_EQ(test.expected_result, actual_result)
+            << fmt::format("ac_type: {}, user_name: {}, database_name: {}",
+                           test.rpc_code,
+                           test.user_name,
+                           test.database_name);
+    }
+}
+
 } // namespace ranger
 } // namespace dsn
diff --git a/src/shell/CMakeLists.txt b/src/shell/CMakeLists.txt
index 6e525d7a0c..70447ed766 100644
--- a/src/shell/CMakeLists.txt
+++ b/src/shell/CMakeLists.txt
@@ -31,6 +31,7 @@ set(MY_PROJ_LIBS
     pegasus_base
     dsn.replication.tool
     dsn_replica_server
+    dsn_meta_server
     dsn_replication_common
     dsn_client
     dsn_utils