diff --git a/src/meta-srv/src/mocks.rs b/src/meta-srv/src/mocks.rs
index 1a6fa306743..1afc5dbd334 100644
--- a/src/meta-srv/src/mocks.rs
+++ b/src/meta-srv/src/mocks.rs
@@ -21,6 +21,7 @@ use api::v1::meta::heartbeat_server::HeartbeatServer;
 use api::v1::meta::procedure_service_server::ProcedureServiceServer;
 use api::v1::meta::store_server::StoreServer;
 use client::client_manager::NodeClients;
+use common_base::Plugins;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_meta::key::TableMetadataManager;
 use common_meta::kv_backend::etcd::EtcdStore;
@@ -95,6 +96,7 @@ pub async fn mock(
         datanode_clients,
         in_memory,
         None,
+        Plugins::default(),
     )
     .await
 }
@@ -114,6 +116,30 @@ pub async fn mock_with_client_channel_config(
         datanode_clients,
         in_memory,
         Some(client_channel_config),
+        Plugins::default(),
+    )
+    .await
+}
+
+/// Builds a mock metasrv with the given [`Plugins`].
+///
+/// Same as [`mock`], except that the plugins are passed to the metasrv builder.
+pub async fn mock_with_plugins(
+    opts: MetasrvOptions,
+    kv_backend: KvBackendRef,
+    selector: Option<SelectorRef>,
+    datanode_clients: Option<Arc<NodeClients>>,
+    in_memory: Option<ResettableKvBackendRef>,
+    plugins: Plugins,
+) -> MockInfo {
+    mock_inner(
+        opts,
+        kv_backend,
+        selector,
+        datanode_clients,
+        in_memory,
+        None,
+        plugins,
     )
     .await
 }
@@ -125,6 +151,7 @@ async fn mock_inner(
     datanode_clients: Option<Arc<NodeClients>>,
     in_memory: Option<ResettableKvBackendRef>,
     client_channel_config: Option<ChannelConfig>,
+    plugins: Plugins,
 ) -> MockInfo {
     let server_addr = opts.grpc.server_addr.clone();
     let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone()));
@@ -152,6 +179,8 @@ async fn mock_inner(
         None => builder,
     };
 
+    let builder = builder.plugins(plugins);
+
     let metasrv = builder.build().await.unwrap();
     metasrv.try_start().await.unwrap();
 
diff --git a/tests-integration/src/cluster.rs b/tests-integration/src/cluster.rs
index 21a0d5eba2d..76f198c9707 100644
--- a/tests-integration/src/cluster.rs
+++ b/tests-integration/src/cluster.rs
@@ -174,6 +174,7 @@ pub struct GreptimeDbClusterBuilder {
     frontend_auto_create_table: bool,
     shared_home_dir: Option<Arc<TempDir>>,
     meta_selector: Option<SelectorRef>,
+    plugins: Plugins,
     local_file_access: LocalFileAccess,
     event_recorder_options: EventRecorderOptions,
 }
@@ -210,6 +211,7 @@ impl GreptimeDbClusterBuilder {
             frontend_auto_create_table: true,
             shared_home_dir: None,
             meta_selector: None,
+            plugins: Plugins::default(),
             local_file_access: LocalFileAccess::default(),
             event_recorder_options: EventRecorderOptions::default(),
         }
@@ -284,6 +286,13 @@ impl GreptimeDbClusterBuilder {
         self
     }
 
+    /// Sets the [Plugins] used by the metasrv.
+    #[must_use]
+    pub fn with_plugins(mut self, plugins: Plugins) -> Self {
+        self.plugins = plugins;
+        self
+    }
+
     /// Configure the frontend COPY sandbox for filesystem integration tests.
     pub fn with_local_file_access(mut self, access: LocalFileAccess) -> Self {
         self.local_file_access = access;
@@ -297,6 +306,7 @@ impl GreptimeDbClusterBuilder {
         guards: Vec<TestGuard>,
     ) -> GreptimeDbCluster {
         let datanodes = datanode_options.len();
+
         let channel_config = ChannelConfig::new().timeout(Some(Duration::from_secs(20)));
         let datanode_clients = Arc::new(NodeClients::new(channel_config));
 
@@ -322,12 +332,13 @@ impl GreptimeDbClusterBuilder {
         test_util::prepare_another_catalog_and_schema_with_kv_backend(self.kv_backend.clone())
             .await;
 
-        let metasrv = meta_srv::mocks::mock(
+        let metasrv = meta_srv::mocks::mock_with_plugins(
             opt,
             self.kv_backend.clone(),
             self.meta_selector.clone(),
             Some(datanode_clients.clone()),
             None,
+            self.plugins.clone(),
         )
         .await;
 
@@ -692,6 +703,8 @@ async fn create_datanode_client(datanode: &Datanode) -> (String, Client) {
 #[cfg(test)]
 #[allow(deprecated)]
 mod tests {
+    use std::sync::atomic::{AtomicBool, Ordering};
+
     use api::v1::flow::FlowRequest;
     use api::v1::region::{
         ListMetadataRequest, RegionRequest, RegionRequestHeader, region_request,
@@ -700,12 +713,14 @@ mod tests {
     use client::Client;
     use common_error::ext::ErrorExt;
     use common_error::status_code::StatusCode;
+    use common_meta::ddl_manager::{DdlManager, DdlManagerConfigurator, DdlManagerConfiguratorRef};
     use common_meta::key::TableMetadataManager;
     use common_meta::key::flow::FlowMetadataManager;
     use common_meta::kv_backend::memory::MemoryKvBackend;
     use common_meta::node_manager::{DatanodeManager, FlownodeManager};
     use common_meta::peer::Peer;
     use flow::{FlownodeBuilder, FlownodeOptions, FlownodeServiceBuilder, FrontendClient};
+    use meta_srv::metasrv::builder::DdlManagerConfigureContext;
 
     use super::*;
 
@@ -882,6 +897,38 @@ mod tests {
         flownode
     }
 
+    struct TestDdlManagerConfigurator(Arc<AtomicBool>);
+
+    #[async_trait::async_trait]
+    impl DdlManagerConfigurator<DdlManagerConfigureContext> for TestDdlManagerConfigurator {
+        async fn configure(
+            &self,
+            ddl_manager: DdlManager,
+            _ctx: DdlManagerConfigureContext,
+        ) -> std::result::Result<DdlManager, common_error::ext::BoxedError> {
+            self.0.store(true, Ordering::SeqCst);
+            Ok(ddl_manager)
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_build_cluster_with_plugins() {
+        let configured = Arc::new(AtomicBool::new(false));
+        let plugins = Plugins::default();
+        plugins.insert::<DdlManagerConfiguratorRef<DdlManagerConfigureContext>>(Arc::new(
+            TestDdlManagerConfigurator(configured.clone()),
+        ));
+
+        let _cluster = GreptimeDbClusterBuilder::new("test_cluster_with_plugins")
+            .await
+            .with_datanodes(1)
+            .with_plugins(plugins)
+            .build(false)
+            .await;
+
+        assert!(configured.load(Ordering::SeqCst));
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_region_requester_health_check_and_list_metadata() {
         // Arrange: configure active and inactive synthetic peers over the duplex harness.