Repository: robustmq/robustmq (language: rust)

Issue:
RBIP-07：Add features for connection jitter detection
https://docs.emqx.com/zh/emqx/latest/access-control/flapping-detect.html


Discussion / hints:
Please assign me. Thanks.

Fix patch:
diff --git a/config/log-config/mqtt-log4rs.yaml b/config/log-config/mqtt-log4rs.yaml
index f3be48be0..90a73a9a7 100644
--- a/config/log-config/mqtt-log4rs.yaml
+++ b/config/log-config/mqtt-log4rs.yaml
@@ -49,7 +49,7 @@ appenders:
         count: 50
 
 root:
-  level: info
+  level: debug
   appenders:
     - stdout
     - server
diff --git a/docs/.design/logic/flapping-detect.md b/docs/.design/logic/flapping-detect.md
index 6aad733ad..c7ff02d6e 100644
--- a/docs/.design/logic/flapping-detect.md
+++ b/docs/.design/logic/flapping-detect.md
@@ -2,29 +2,24 @@
 构建一个结构体
 create a struct
 ```rust
-struct ClientConnectInfo {
+struct FlappingDetectCondition {
     client_id: u64,
-    connect_count: u8,
-    request_time: u64,
+    connect_times: u8,
+    first_request_time: u64,
 }
 ```
 
-创建或获取当前的counter
+获取当前的时间作为最近请求时间`current_request_time`
+获取当前的请求次数`current_counter`
 
-counter值增加1
+从Map中获取对应`client-id`的结构信息`FlappingDetectCondition`，如果没有的情况下
+我们需要自己构建`FlappingDetectCondition`, 并将`FlappingDetectCondition`放入map中
 
-获取当前时间`now_time`作为`current_request_time`
+如果`current_request_time` - `request_time` < `windows_time` 并且
+`counter` - `connect_count` > `max_count`：
+ 将对应的client_id放入到黑名单当中，并指定封禁时间
+> 这里的两个校验分别是：是否在窗口时间内以及是否超过窗口时间内的最大连接次数
 
-从Map中获取对应`client-id`的结构信息`ClientConnectInfo`
-
-如果`current_request_time` - `request_time` < `windows_time`
-- 如果`counter` - `connect_count` > `max_count`：
- 将对应的client放入到黑名单当中，并指定封禁时间
-- 否则只更新connect_count放入到map中
-
-更新为本次request_time和connect_count
-
-重新放入map中
 
 # 定时清理map
 ## 目的
@@ -32,6 +27,5 @@ counter值增加1
 
 ## 核心逻辑
 
-获取now_time
-
-如果now_time - request_time > windows_time, 删除map中的元素
+线程会在窗口时间触发后执行，主要的执行逻辑如下：
+清理时间和首次请求时间已经大于当前的时间窗口了，此时就可以进行清理操作
diff --git a/docs/.design/storage-adapter/rocksdb.md b/docs/.design/storage-adapter/rocksdb.md
index 002b029af..d41ff9a2c 100644
--- a/docs/.design/storage-adapter/rocksdb.md
+++ b/docs/.design/storage-adapter/rocksdb.md
@@ -49,6 +49,6 @@ This key value pair is used for fast record retrieval in `read_by_tag` method. I
 
 key: `/group/{group}/{namespace}/{shard}`
 
-value: offset of shard `shard` under namespace `namespace` in group `group` 
+value: offset of shard `shard` under namespace `namespace` in group `group`
 
 This key value pair will be set in `commit_offset` method
diff --git a/docs/.design/user-draft/flapping-detect.md b/docs/.design/user-draft/flapping-detect.md
index 9c2dabdc9..02d485a2b 100644
--- a/docs/.design/user-draft/flapping-detect.md
+++ b/docs/.design/user-draft/flapping-detect.md
@@ -1,4 +1,4 @@
-# 消息抖动功能用户文档
+# 连接抖动功能用户文档
 ## 功能介绍
 RobustMQ支持自动封禁那些被检测到短时间内频繁登录的客户端，
 并且在一段时间内拒绝这些客户端的登录，
@@ -31,7 +31,7 @@ roubst-cli mqtt flapping-detect --window-time=1
 ```
 - 修改最大断连次数
 ```shell
-roubst-cli mqtt flapping-detect --max-disconnects=15
+roubst-cli mqtt flapping-detect --max-client-connections=15
 ```
 - 修改封禁时长
 ```shell
diff --git a/src/cli-command/src/mqtt.rs b/src/cli-command/src/mqtt.rs
index 317f7cae6..1ef742054 100644
--- a/src/cli-command/src/mqtt.rs
+++ b/src/cli-command/src/mqtt.rs
@@ -14,23 +14,26 @@
 
 use crate::template::{PublishArgsRequest, SubscribeArgsRequest};
 use crate::{connect_server5, error_info, grpc_addr};
-use common_base::enum_type::common_enum::SortType;
+use common_base::enum_type::sort_type::SortType;
 use common_base::tools::unique_id;
 use grpc_clients::mqtt::admin::call::{
     cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user,
-    mqtt_broker_enable_slow_subscribe, mqtt_broker_list_connection,
-    mqtt_broker_list_slow_subscribe, mqtt_broker_list_topic, mqtt_broker_list_user,
+    mqtt_broker_enable_flapping_detect, mqtt_broker_enable_slow_subscribe,
+    mqtt_broker_list_connection, mqtt_broker_list_slow_subscribe, mqtt_broker_list_topic,
+    mqtt_broker_list_user,
 };
 use grpc_clients::pool::ClientPool;
 use metadata_struct::mqtt::user::MqttUser;
 use paho_mqtt::{DisconnectOptionsBuilder, MessageBuilder, Properties, PropertyCode, ReasonCode};
 use prettytable::{row, Table};
 use protocol::broker_mqtt::broker_mqtt_admin::{
-    ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, EnableSlowSubscribeRequest,
-    ListConnectionRequest, ListSlowSubscribeRequest, ListTopicRequest, ListUserRequest,
+    ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, EnableFlappingDetectRequest,
+    EnableSlowSubscribeRequest, ListConnectionRequest, ListSlowSubscribeRequest, ListTopicRequest,
+    ListUserRequest,
 };
 use std::str::FromStr;
 use std::sync::Arc;
+
 use tokio::io::{self, AsyncBufReadExt, BufReader};
 use tokio::{select, signal};
 
@@ -52,10 +55,13 @@ pub enum MqttActionType {
     // connection
     ListConnection,
 
-    // observability: slow-ub
+    // observability: slow-sub
     EnableSlowSubscribe(EnableSlowSubscribeRequest),
     ListSlowSubscribe(ListSlowSubscribeRequest),
 
+    // flapping detect
+    EnableFlappingDetect(EnableFlappingDetectRequest),
+
     // publish
     Publish(PublishArgsRequest),
 
@@ -110,6 +116,10 @@ impl MqttBrokerCommand {
                 self.list_slow_subscribe(&client_pool, params.clone(), request.clone())
                     .await;
             }
+            MqttActionType::EnableFlappingDetect(ref request) => {
+                self.enable_flapping_detect(&client_pool, params.clone(), *request)
+                    .await;
+            }
             MqttActionType::Publish(ref request) => {
                 self.publish(params.clone(), request.clone()).await;
             }
@@ -329,6 +339,37 @@ impl MqttBrokerCommand {
         }
     }
 
+    // flapping detect
+    async fn enable_flapping_detect(
+        &self,
+        client_pool: &ClientPool,
+        params: MqttCliCommandParam,
+        cli_request: EnableFlappingDetectRequest,
+    ) {
+        match mqtt_broker_enable_flapping_detect(
+            client_pool,
+            &grpc_addr(params.server),
+            cli_request,
+        )
+        .await
+        {
+            Ok(reply) => {
+                if reply.is_enable {
+                    println!("The flapping detect feature has been successfully enabled.");
+                } else {
+                    println!("The flapping detect feature has been successfully closed.");
+                }
+            }
+
+            Err(e) => {
+                println!(
+                    "The flapping detect feature failed to enable, with the specific reason being:"
+                );
+                error_info(e.to_string());
+            }
+        }
+    }
+
     // ---------------- observability ----------------
     // ------------ slow subscribe features ----------
     async fn enable_slow_subscribe(
diff --git a/src/cmd/src/cli-command/command.rs b/src/cmd/src/cli-command/command.rs
index 041b3c3fd..1f7461280 100644
--- a/src/cmd/src/cli-command/command.rs
+++ b/src/cmd/src/cli-command/command.rs
@@ -20,13 +20,15 @@ use cli_command::placement::{
     PlacementActionType, PlacementCenterCommand, PlacementCliCommandParam,
 };
 use mqtt::publish::process_subscribe_args;
-use protocol::broker_mqtt::broker_mqtt_admin::ListTopicRequest;
+use protocol::broker_mqtt::broker_mqtt_admin::{EnableFlappingDetectRequest, ListTopicRequest};
 
 use protocol::placement_center::placement_center_openraft::{
     AddLearnerRequest, ChangeMembershipRequest, Node,
 };
 
-use crate::mqtt::admin::{process_slow_sub_args, process_user_args, MqttUserCommand, SlowSubArgs};
+use crate::mqtt::admin::{
+    process_slow_sub_args, process_user_args, FlappingDetectArgs, MqttUserCommand, SlowSubArgs,
+};
 use crate::mqtt::publish::{process_publish_args, PubSubArgs};
 
 #[derive(Parser)] // requires `derive` feature
@@ -78,6 +80,10 @@ enum MQTTAction {
     // Connections
     ListConnection,
 
+    // flapping detect feat
+    #[clap(name = "flapping-detect")]
+    FlappingDetect(FlappingDetectArgs),
+
     ListTopic(ListTopicArgs),
 
     Publish(PubSubArgs),
@@ -187,9 +193,18 @@ async fn handle_mqtt(args: MqttArgs, cmd: MqttBrokerCommand) {
                     MatchOption::S => 2,
                 },
             }),
+            MQTTAction::SlowSub(args) => process_slow_sub_args(args),
+            MQTTAction::FlappingDetect(args) => {
+                MqttActionType::EnableFlappingDetect(EnableFlappingDetectRequest {
+                    is_enable: args.is_enable.unwrap_or(false),
+                    window_time: args.window_time.unwrap_or(1),
+                    max_client_connections: args.max_client_connections.unwrap_or(15),
+                    ban_time: args.ban_time.unwrap_or(5),
+                })
+            }
             MQTTAction::Publish(args) => process_publish_args(args),
             MQTTAction::Subscribe(args) => process_subscribe_args(args),
-            MQTTAction::SlowSub(args) => process_slow_sub_args(args), // _ => unreachable!("UnSupport command"),
+            // _ => unreachable!("UnSupport command"),
         },
     };
     cmd.start(params).await;
diff --git a/src/cmd/src/cli-command/mqtt/admin.rs b/src/cmd/src/cli-command/mqtt/admin.rs
index 5f4694db9..fbb0d79b4 100644
--- a/src/cmd/src/cli-command/mqtt/admin.rs
+++ b/src/cmd/src/cli-command/mqtt/admin.rs
@@ -17,9 +17,9 @@ use core::option::Option::Some;
 use clap::builder::{
     ArgAction, BoolishValueParser, EnumValueParser, NonEmptyStringValueParser, RangedU64ValueParser,
 };
-use clap::Parser;
+use clap::{arg, Parser};
 use cli_command::mqtt::MqttActionType;
-use common_base::enum_type::common_enum::SortType;
+use common_base::enum_type::sort_type::SortType;
 use protocol::broker_mqtt::broker_mqtt_admin::{CreateUserRequest, DeleteUserRequest};
 use protocol::broker_mqtt::broker_mqtt_admin::{
     EnableSlowSubscribeRequest, ListSlowSubscribeRequest,
@@ -64,6 +64,38 @@ pub(crate) struct DeleteUserArgs {
     pub(crate) username: String,
 }
 
+// flapping detect feat
+#[derive(Debug, Parser)]
+#[command(author="RobustMQ", about="", long_about = None)]
+#[command(next_line_help = true)]
+pub(crate) struct FlappingDetectArgs {
+    #[arg(long = "enable")]
+    #[arg(value_parser =  BoolishValueParser::new())]
+    #[arg(default_missing_value = "false")]
+    #[arg(action = ArgAction::Set, num_args = 0..=1)]
+    #[arg(required = true, require_equals = true, exclusive = true)]
+    #[arg(help = "Enable or disable the feature")]
+    pub(crate) is_enable: Option<bool>,
+    #[arg(long = "window-time")]
+    #[arg(value_parser = RangedU64ValueParser::<u32>::new())]
+    #[arg(default_missing_value = "1")]
+    #[arg(action = ArgAction::Set, num_args = 0..=1)]
+    #[arg(require_equals = true)]
+    #[arg(help = "unit is minutes")]
+    pub(crate) window_time: Option<u32>,
+    #[arg(long = "max-client-connections")]
+    #[arg(value_parser = RangedU64ValueParser::<u32>::new())]
+    #[arg(default_missing_value = "15")]
+    #[arg(action = ArgAction::Set, num_args = 0..=1)]
+    pub(crate) max_client_connections: Option<u32>,
+    #[arg(long = "ban-time")]
+    #[arg(value_parser = RangedU64ValueParser::<u32>::new())]
+    #[arg(default_missing_value = "5")]
+    #[arg(action = ArgAction::Set, num_args = 0..=1)]
+    #[arg(help = "unit is minutes")]
+    pub(crate) ban_time: Option<u32>,
+}
+
 // observability: slow-sub feat
 #[derive(Debug, Parser)]
 #[command(author="RobustMQ", about="", long_about = None)]
@@ -76,30 +108,30 @@ pub(crate) struct SlowSubArgs {
     #[arg(conflicts_with_all = ["list", "sort", "topic", "sub_name", "client_id"])]
     #[arg(help = "Enable or disable the feature")]
     is_enable: Option<bool>,
-    #[arg(long)]
+    #[arg(long = "list")]
     #[arg(value_parser = RangedU64ValueParser::<u64>::new())]
     #[arg(default_missing_value = "100")]
     #[arg(action = ArgAction::Set, num_args = 0..=1)]
     #[arg(require_equals = true)]
     list: Option<u64>,
-    #[arg(long)]
+    #[arg(long = "sort")]
     #[arg(required = false, requires = "list", require_equals = true)]
     #[arg(value_parser = EnumValueParser::<SortType>::new())]
     #[arg(default_missing_value = "DESC", ignore_case = true)]
     #[arg(action = ArgAction::Set, num_args = 0..=1)]
     #[arg(help = "Sort the results")]
     sort: Option<SortType>,
-    #[arg(long)]
+    #[arg(long = "topic")]
     #[arg(required = false, requires = "list", require_equals = true)]
     #[arg(action = ArgAction::Set, num_args = 0..=1)]
     #[arg(value_parser = NonEmptyStringValueParser::new())]
     topic: Option<String>,
-    #[arg(long)]
+    #[arg(long = "sub-name")]
     #[arg(required = false, requires = "list", require_equals = true)]
     #[arg(action = ArgAction::Set, num_args = 0..=1)]
     #[arg(value_parser = NonEmptyStringValueParser::new())]
     sub_name: Option<String>,
-    #[arg(long)]
+    #[arg(long = "client-id")]
     #[arg(required = false, requires = "list", require_equals = true)]
     #[arg(action = ArgAction::Set, num_args = 0..=1)]
     #[arg(value_parser = NonEmptyStringValueParser::new())]
diff --git a/src/common/base/src/enum_type/mod.rs b/src/common/base/src/enum_type/mod.rs
index a1df20a37..d6550d7c3 100644
--- a/src/common/base/src/enum_type/mod.rs
+++ b/src/common/base/src/enum_type/mod.rs
@@ -12,5 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod common_enum;
+pub mod sort_type;
+pub mod time_unit_enum;
 pub mod topic_rewrite_action_enum;
diff --git a/src/common/base/src/enum_type/common_enum.rs b/src/common/base/src/enum_type/sort_type.rs
similarity index 100%
rename from src/common/base/src/enum_type/common_enum.rs
rename to src/common/base/src/enum_type/sort_type.rs
diff --git a/src/common/base/src/enum_type/time_unit_enum.rs b/src/common/base/src/enum_type/time_unit_enum.rs
new file mode 100644
index 000000000..64866c5ab
--- /dev/null
+++ b/src/common/base/src/enum_type/time_unit_enum.rs
@@ -0,0 +1,64 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt;
+use std::fmt::Formatter;
+use std::str::FromStr;
+
+use clap::builder::PossibleValue;
+use clap::ValueEnum;
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum TimeUnit {
+    Seconds,
+    Minutes,
+    Hours,
+    Days,
+}
+
+impl FromStr for TimeUnit {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        for variant in Self::value_variants() {
+            if variant.to_possible_value().unwrap().matches(s, false) {
+                return Ok(*variant);
+            }
+        }
+        Err(format!("invalid variant: {s}"))
+    }
+}
+
+impl fmt::Display for TimeUnit {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        self.to_possible_value()
+            .expect("no values are skipped")
+            .get_name()
+            .fmt(f)
+    }
+}
+impl ValueEnum for TimeUnit {
+    fn value_variants<'a>() -> &'a [Self] {
+        &[Self::Minutes, Self::Hours, Self::Days]
+    }
+
+    fn to_possible_value(&self) -> Option<PossibleValue> {
+        Some(match self {
+            TimeUnit::Minutes => PossibleValue::new("minutes"),
+            TimeUnit::Hours => PossibleValue::new("hours"),
+            TimeUnit::Days => PossibleValue::new("days"),
+            TimeUnit::Seconds => PossibleValue::new("seconds"),
+        })
+    }
+}
diff --git a/src/common/base/src/tools.rs b/src/common/base/src/tools.rs
index b61982e64..a8408e144 100644
--- a/src/common/base/src/tools.rs
+++ b/src/common/base/src/tools.rs
@@ -21,6 +21,7 @@ use log::warn;
 use tonic::Status;
 use uuid::Uuid;
 
+use crate::enum_type::time_unit_enum::TimeUnit;
 use crate::error::common::CommonError;
 
 /// Create multi-level directory
@@ -81,6 +82,17 @@ pub fn unique_id() -> String {
     uuid.to_string().replace("-", "")
 }
 
+pub fn convert_seconds(number: u64, unit: TimeUnit) -> u64 {
+    if unit == TimeUnit::Minutes {
+        return number * 60;
+    } else if unit == TimeUnit::Hours {
+        return number * 3600;
+    } else if unit == TimeUnit::Days {
+        return number * 86400;
+    }
+    number
+}
+
 /// Obtain local IP address
 ///
 /// This function attempts to obtain the local IP address of the device and returns the address as a string upon success
@@ -153,7 +165,8 @@ where
 
 #[cfg(test)]
 mod tests {
-    use crate::tools::{get_local_ip, unique_id};
+    use crate::enum_type::time_unit_enum::TimeUnit;
+    use crate::tools::{convert_seconds, get_local_ip, unique_id};
 
     #[test]
     fn get_local_ip_test() {
@@ -164,4 +177,9 @@ mod tests {
     fn unique_id_test() {
         println!("{}", unique_id());
     }
+
+    #[test]
+    fn test_convert_seconds() {
+        assert_eq!(convert_seconds(1, TimeUnit::Minutes), 60);
+    }
 }
diff --git a/src/common/metadata-struct/src/mqtt/cluster.rs b/src/common/metadata-struct/src/mqtt/cluster.rs
index 39774420a..b0813b4a2 100644
--- a/src/common/metadata-struct/src/mqtt/cluster.rs
+++ b/src/common/metadata-struct/src/mqtt/cluster.rs
@@ -23,6 +23,7 @@ pub struct MqttClusterDynamicConfig {
     pub security: MqttClusterDynamicConfigSecurity,
     pub network: MqttClusterDynamicConfigNetwork,
     pub slow: MqttClusterDynamicSlowSub,
+    pub flapping_detect: MqttClusterDynamicFlappingDetect,
 }
 
 // MQTT cluster protocol related dynamic configuration
@@ -75,6 +76,14 @@ pub struct MqttClusterDynamicSlowSub {
     pub response_ms: u32,
 }
 
+#[derive(Serialize, Deserialize, Default, Clone)]
+pub struct MqttClusterDynamicFlappingDetect {
+    pub enable: bool,
+    pub window_time: u32,
+    pub max_client_connections: u64,
+    pub ban_time: u32,
+}
+
 impl MqttClusterDynamicConfig {
     pub fn new() -> Self {
         MqttClusterDynamicConfig {
@@ -114,6 +123,12 @@ impl MqttClusterDynamicConfig {
                 internal_ms: 0,
                 response_ms: 0,
             },
+            flapping_detect: MqttClusterDynamicFlappingDetect {
+                enable: false,
+                window_time: 1,
+                max_client_connections: 15,
+                ban_time: 5,
+            },
         }
     }
 
diff --git a/src/grpc-clients/src/mqtt/admin/call.rs b/src/grpc-clients/src/mqtt/admin/call.rs
index 9444f54d1..ec256cee9 100644
--- a/src/grpc-clients/src/mqtt/admin/call.rs
+++ b/src/grpc-clients/src/mqtt/admin/call.rs
@@ -18,10 +18,11 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
     CreateBlacklistReply, CreateBlacklistRequest, CreateTopicRewriteRuleReply,
     CreateTopicRewriteRuleRequest, CreateUserReply, CreateUserRequest, DeleteAclReply,
     DeleteAclRequest, DeleteBlacklistReply, DeleteBlacklistRequest, DeleteTopicRewriteRuleReply,
-    DeleteTopicRewriteRuleRequest, DeleteUserReply, DeleteUserRequest, EnableSlowSubScribeReply,
-    EnableSlowSubscribeRequest, ListAclReply, ListAclRequest, ListBlacklistReply,
-    ListBlacklistRequest, ListConnectionReply, ListConnectionRequest, ListSlowSubscribeReply,
-    ListSlowSubscribeRequest, ListTopicReply, ListTopicRequest, ListUserReply, ListUserRequest,
+    DeleteTopicRewriteRuleRequest, DeleteUserReply, DeleteUserRequest, EnableFlappingDetectReply,
+    EnableFlappingDetectRequest, EnableSlowSubScribeReply, EnableSlowSubscribeRequest,
+    ListAclReply, ListAclRequest, ListBlacklistReply, ListBlacklistRequest, ListConnectionReply,
+    ListConnectionRequest, ListSlowSubscribeReply, ListSlowSubscribeRequest, ListTopicReply,
+    ListTopicRequest, ListUserReply, ListUserRequest,
 };
 
 use crate::pool::ClientPool;
@@ -119,6 +120,15 @@ pub async fn mqtt_broker_list_connection(
     retry_call(client_pool, addrs, request).await
 }
 
+// -------flapping detect feat  -----------
+pub async fn mqtt_broker_enable_flapping_detect(
+    client_pool: &ClientPool,
+    addrs: &[impl AsRef<str>],
+    request: EnableFlappingDetectRequest,
+) -> Result<EnableFlappingDetectReply, CommonError> {
+    retry_call(client_pool, addrs, request).await
+}
+
 // --------- observability --------
 // --------- slow subscribe features ------
 pub async fn mqtt_broker_enable_slow_subscribe(
diff --git a/src/grpc-clients/src/mqtt/mod.rs b/src/grpc-clients/src/mqtt/mod.rs
index 753312907..b2a4a2149 100644
--- a/src/grpc-clients/src/mqtt/mod.rs
+++ b/src/grpc-clients/src/mqtt/mod.rs
@@ -18,10 +18,11 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
     CreateBlacklistReply, CreateBlacklistRequest, CreateTopicRewriteRuleReply,
     CreateTopicRewriteRuleRequest, CreateUserReply, CreateUserRequest, DeleteAclReply,
     DeleteAclRequest, DeleteBlacklistReply, DeleteBlacklistRequest, DeleteTopicRewriteRuleReply,
-    DeleteTopicRewriteRuleRequest, DeleteUserReply, DeleteUserRequest, EnableSlowSubScribeReply,
-    EnableSlowSubscribeRequest, ListAclReply, ListAclRequest, ListBlacklistReply,
-    ListBlacklistRequest, ListConnectionReply, ListConnectionRequest, ListSlowSubscribeReply,
-    ListSlowSubscribeRequest, ListTopicReply, ListTopicRequest, ListUserReply, ListUserRequest,
+    DeleteTopicRewriteRuleRequest, DeleteUserReply, DeleteUserRequest, EnableFlappingDetectReply,
+    EnableFlappingDetectRequest, EnableSlowSubScribeReply, EnableSlowSubscribeRequest,
+    ListAclReply, ListAclRequest, ListBlacklistReply, ListBlacklistRequest, ListConnectionReply,
+    ListConnectionRequest, ListSlowSubscribeReply, ListSlowSubscribeRequest, ListTopicReply,
+    ListTopicRequest, ListUserReply, ListUserRequest,
 };
 use protocol::broker_mqtt::broker_mqtt_inner::mqtt_broker_inner_service_client::MqttBrokerInnerServiceClient;
 use protocol::broker_mqtt::broker_mqtt_inner::{
@@ -147,6 +148,14 @@ impl_retriable_request!(
     mqtt_broker_list_connection
 );
 
+impl_retriable_request!(
+    EnableFlappingDetectRequest,
+    MqttBrokerAdminServiceClient<Channel>,
+    EnableFlappingDetectReply,
+    mqtt_broker_admin_services_client,
+    mqtt_broker_enable_flapping_detect
+);
+
 impl_retriable_request!(
     EnableSlowSubscribeRequest,
     MqttBrokerAdminServiceClient<Channel>,
diff --git a/src/mqtt-broker/src/handler/cluster_config.rs b/src/mqtt-broker/src/handler/cluster_config.rs
index 76043e23c..eb5f1ad08 100644
--- a/src/mqtt-broker/src/handler/cluster_config.rs
+++ b/src/mqtt-broker/src/handler/cluster_config.rs
@@ -12,17 +12,55 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use metadata_struct::mqtt::cluster::{MqttClusterDynamicConfig, MqttClusterDynamicSlowSub};
-
 use crate::handler::cache::CacheManager;
 use crate::handler::error::MqttBrokerError;
 use crate::storage::cluster::ClusterStorage;
+use metadata_struct::mqtt::cluster::{
+    MqttClusterDynamicConfig, MqttClusterDynamicFlappingDetect, MqttClusterDynamicSlowSub,
+};
 
 /// This section primarily implements cache management for cluster-related configuration operations.
 /// Through this implementation, we can retrieve configuration information within the cluster
 /// and set corresponding cluster configuration attributes.
 impl CacheManager {
-    pub(super) fn set_cluster_info(&self, cluster: MqttClusterDynamicConfig) {
+    pub async fn set_flapping_detect_config(
+        &self,
+        config: MqttClusterDynamicFlappingDetect,
+    ) -> Result<(), MqttBrokerError> {
+        let mut dyn_config = self.get_cluster_info();
+        dyn_config.flapping_detect = config;
+
+        // save in cache
+        self.set_cluster_info(dyn_config.clone());
+
+        // save in storage
+        self.save_dynamic_config(dyn_config).await?;
+        Ok(())
+    }
+    pub fn get_flapping_detect_config(&self) -> MqttClusterDynamicFlappingDetect {
+        self.get_cluster_info().flapping_detect
+    }
+
+    pub async fn set_slow_sub_config(
+        &self,
+        config: MqttClusterDynamicSlowSub,
+    ) -> Result<(), MqttBrokerError> {
+        let mut dynamic_config = self.get_cluster_info();
+        dynamic_config.slow = config;
+
+        // save in cache
+        self.set_cluster_info(dynamic_config.clone());
+
+        // save in storage
+        self.save_dynamic_config(dynamic_config).await?;
+        Ok(())
+    }
+
+    pub fn get_slow_sub_config(&self) -> MqttClusterDynamicSlowSub {
+        self.get_cluster_info().slow
+    }
+
+    pub fn set_cluster_info(&self, cluster: MqttClusterDynamicConfig) {
         self.cluster_info.insert(self.cluster_name.clone(), cluster);
     }
 
@@ -33,13 +71,10 @@ impl CacheManager {
         MqttClusterDynamicConfig::new()
     }
 
-    pub async fn enable_slow_sub(&self, is_enable: bool) -> Result<(), MqttBrokerError> {
-        // save in cache
-        let mut dynamic_config = self.get_cluster_info();
-        dynamic_config.slow.enable = is_enable;
-        self.set_cluster_info(dynamic_config.clone());
-
-        // save in storage
+    async fn save_dynamic_config(
+        &self,
+        dynamic_config: MqttClusterDynamicConfig,
+    ) -> Result<(), MqttBrokerError> {
         let client_pool = self.client_pool.clone();
         let cluster_storage = ClusterStorage::new(client_pool);
         cluster_storage
@@ -47,8 +82,4 @@ impl CacheManager {
             .await?;
         Ok(())
     }
-
-    pub fn get_slow_sub_config(&self) -> MqttClusterDynamicSlowSub {
-        self.get_cluster_info().slow
-    }
 }
diff --git a/src/mqtt-broker/src/handler/command.rs b/src/mqtt-broker/src/handler/command.rs
index 97a12c79b..59a8cd4b1 100644
--- a/src/mqtt-broker/src/handler/command.rs
+++ b/src/mqtt-broker/src/handler/command.rs
@@ -113,7 +113,7 @@ where
                 connect,
                 properties,
                 last_will,
-                last_will_peoperties,
+                last_will_properties,
                 login,
             ) => {
                 connect_manager
@@ -127,7 +127,7 @@ where
                                 connect,
                                 properties,
                                 last_will,
-                                last_will_peoperties,
+                                last_will_properties,
                                 &login,
                                 addr,
                             )
@@ -141,7 +141,7 @@ where
                                 connect,
                                 properties,
                                 last_will,
-                                last_will_peoperties,
+                                last_will_properties,
                                 &login,
                                 addr,
                             )
@@ -155,7 +155,7 @@ where
                                 connect,
                                 properties,
                                 last_will,
-                                last_will_peoperties,
+                                last_will_properties,
                                 &login,
                                 addr,
                             )
diff --git a/src/mqtt-broker/src/handler/flapping_detect.rs b/src/mqtt-broker/src/handler/flapping_detect.rs
new file mode 100644
index 000000000..4a04b5ceb
--- /dev/null
+++ b/src/mqtt-broker/src/handler/flapping_detect.rs
@@ -0,0 +1,186 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::handler::cache::CacheManager;
+use crate::handler::error::MqttBrokerError;
+use crate::observability::metrics::event_metrics;
+use common_base::enum_type::time_unit_enum::TimeUnit;
+use common_base::tools::{convert_seconds, now_second};
+use log::{debug, error, info};
+use metadata_struct::acl::mqtt_blacklist::{MqttAclBlackList, MqttAclBlackListType};
+use metadata_struct::mqtt::cluster::MqttClusterDynamicFlappingDetect;
+use protocol::broker_mqtt::broker_mqtt_admin::EnableFlappingDetectRequest;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::select;
+use tokio::sync::broadcast;
+use tokio::time::sleep;
+
+pub struct UpdateFlappingDetectCache {
+    stop_send: broadcast::Sender<bool>,
+    cache_manager: Arc<CacheManager>,
+}
+
+#[derive(Clone, Debug)]
+pub struct FlappingDetectCondition {
+    pub client_id: String,
+    pub connect_times: u64,
+    pub first_request_time: u64,
+}
+
+impl UpdateFlappingDetectCache {
+    pub fn new(stop_send: broadcast::Sender<bool>, cache_manager: Arc<CacheManager>) -> Self {
+        Self {
+            stop_send,
+            cache_manager,
+        }
+    }
+
+    pub async fn start_update(&self) {
+        loop {
+            let mut stop_rx = self.stop_send.subscribe();
+            select! {
+                val = stop_rx.recv() =>{
+                    if let Ok(flag) = val {
+                        if flag {
+                            info!("{}","Flapping detect cache updating thread stopped successfully.");
+                            break;
+                        }
+                    }
+                }
+                _ = self.update_flapping_detect_cache()=>{
+                }
+            }
+        }
+    }
+
+    async fn update_flapping_detect_cache(&self) {
+        let config = self.cache_manager.get_flapping_detect_config().clone();
+        let window_time = config.window_time as u64;
+        let window_time_2_seconds = convert_seconds(window_time, TimeUnit::Minutes);
+        match self
+            .cache_manager
+            .acl_metadata
+            .remove_flapping_detect_conditions(config)
+            .await
+        {
+            Ok(_) => {
+                info!("Updating Flapping detect cache norm exception");
+            }
+            Err(e) => {
+                error!("{}", e);
+            }
+        }
+        sleep(Duration::from_secs(window_time_2_seconds)).await;
+    }
+}
+
+pub fn check_flapping_detect(client_id: String, cache_manager: &Arc<CacheManager>) {
+    // get metric
+    let current_counter = event_metrics::get_client_connection_counter(client_id.clone());
+    let current_request_time = now_second();
+
+    // get flapping detect info
+    let flapping_detect_condition = if let Some(flapping_detect_info) = cache_manager
+        .acl_metadata
+        .get_flapping_detect_condition(client_id.clone())
+    {
+        flapping_detect_info
+    } else {
+        FlappingDetectCondition {
+            client_id: client_id.clone(),
+            connect_times: current_counter,
+            first_request_time: current_request_time,
+        }
+    };
+
+    debug!(
+        "get a flapping_detect_condition: {:?}",
+        flapping_detect_condition.clone()
+    );
+
+    // incr metric
+    event_metrics::incr_client_connection_counter(client_id.clone());
+
+    let config = cache_manager.get_flapping_detect_config();
+    let current_counter = event_metrics::get_client_connection_counter(client_id.clone());
+
+    if is_within_window_time(
+        current_request_time,
+        flapping_detect_condition.first_request_time,
+        config.window_time as u64,
+    ) && is_exceed_max_client_connections(
+        current_counter,
+        flapping_detect_condition.connect_times,
+        config.max_client_connections,
+    ) {
+        debug!("add a new client_id: {client_id} into blacklist.");
+        add_blacklist_4_connection_jitter(cache_manager, config, client_id);
+    }
+
+    cache_manager
+        .acl_metadata
+        .add_flapping_detect_condition(flapping_detect_condition);
+}
+
+fn add_blacklist_4_connection_jitter(
+    cache_manager: &Arc<CacheManager>,
+    config: MqttClusterDynamicFlappingDetect,
+    client_id: String,
+) {
+    let client_id_blacklist = MqttAclBlackList {
+        blacklist_type: MqttAclBlackListType::ClientId,
+        resource_name: client_id,
+        end_time: now_second() + convert_seconds(config.ban_time as u64, TimeUnit::Minutes),
+        desc: "Ban due to connection jitter ".to_string(),
+    };
+
+    cache_manager.add_blacklist(client_id_blacklist);
+}
+
+fn is_within_window_time(
+    current_request_time: u64,
+    first_request_time: u64,
+    window_time: u64,
+) -> bool {
+    let window_time_seconds = convert_seconds(window_time, TimeUnit::Minutes);
+
+    current_request_time - first_request_time < window_time_seconds
+}
+
+fn is_exceed_max_client_connections(
+    current_time: u64,
+    connect_times: u64,
+    max_client_connections: u64,
+) -> bool {
+    current_time - connect_times > max_client_connections
+}
+
+pub async fn enable_flapping_detect(
+    cache_manager: &Arc<CacheManager>,
+    request: EnableFlappingDetectRequest,
+) -> Result<(), MqttBrokerError> {
+    let connection_jitter = MqttClusterDynamicFlappingDetect {
+        enable: request.is_enable,
+        window_time: request.window_time,
+        max_client_connections: request.max_client_connections as u64,
+        ban_time: request.ban_time,
+    };
+
+    cache_manager
+        .set_flapping_detect_config(connection_jitter)
+        .await?;
+
+    Ok(())
+}
diff --git a/src/mqtt-broker/src/handler/mod.rs b/src/mqtt-broker/src/handler/mod.rs
index 64a5b1a65..af69f3ca5 100644
--- a/src/mqtt-broker/src/handler/mod.rs
+++ b/src/mqtt-broker/src/handler/mod.rs
@@ -19,6 +19,7 @@ pub mod command;
 pub mod connection;
 pub mod constant;
 pub mod error;
+pub mod flapping_detect;
 pub mod flow_control;
 pub mod heartbreat;
 pub mod keep_alive;
diff --git a/src/mqtt-broker/src/handler/mqtt.rs b/src/mqtt-broker/src/handler/mqtt.rs
index 9e9a02e92..59d978c51 100644
--- a/src/mqtt-broker/src/handler/mqtt.rs
+++ b/src/mqtt-broker/src/handler/mqtt.rs
@@ -37,6 +37,7 @@ use crate::handler::cache::{
     CacheManager, ConnectionLiveTime, QosAckPackageData, QosAckPackageType,
 };
 use crate::handler::connection::{build_connection, get_client_id};
+use crate::handler::flapping_detect::check_flapping_detect;
 use crate::handler::lastwill::save_last_will_message;
 use crate::handler::pkid::{pkid_delete, pkid_exists, pkid_save};
 use crate::handler::response::{
@@ -152,6 +153,10 @@ where
             }
         }
 
+        if cluster.flapping_detect.enable {
+            check_flapping_detect(connect.client_id.clone(), &self.cache_manager);
+        }
+
         let (client_id, new_client_id) = get_client_id(&connect.client_id);
 
         let connection = build_connection(
@@ -163,6 +168,15 @@ where
             &addr,
         );
 
+        if self.auth_driver.allow_connect(&connection).await {
+            return response_packet_mqtt_connect_fail(
+                &self.protocol,
+                ConnectReturnCode::Banned,
+                &connect_properties,
+                None,
+            );
+        }
+
         let (session, new_session) = match build_session(
             connect_id,
             client_id.clone(),
diff --git a/src/mqtt-broker/src/handler/validator.rs b/src/mqtt-broker/src/handler/validator.rs
index 73e5f1eeb..dc95e40c5 100644
--- a/src/mqtt-broker/src/handler/validator.rs
+++ b/src/mqtt-broker/src/handler/validator.rs
@@ -12,11 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::min;
-use std::net::SocketAddr;
-use std::sync::Arc;
-
-use futures::SinkExt;
+use futures_util::SinkExt;
 use grpc_clients::pool::ClientPool;
 use log::error;
 use metadata_struct::mqtt::cluster::MqttClusterDynamicConfig;
@@ -27,6 +23,11 @@ use protocol::mqtt::common::{
     LastWillProperties, Login, MqttPacket, MqttProtocol, PubAckReason, PubRecReason, Publish,
     PublishProperties, QoS, Subscribe, SubscribeReasonCode, UnsubAckReason, Unsubscribe,
 };
+use std::cmp::min;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::io::{AsyncWrite, AsyncWriteExt, WriteHalf};
+use tokio::net::TcpStream;
 use tokio_util::codec::FramedWrite;
 
 use super::cache::CacheManager;
@@ -49,56 +50,16 @@ use crate::subscribe::sub_common::sub_path_validator;
 pub async fn tcp_establish_connection_check(
     addr: &SocketAddr,
     connection_manager: &Arc<ConnectionManager>,
-    write_frame_stream: &mut FramedWrite<tokio::io::WriteHalf<tokio::net::TcpStream>, MqttCodec>,
+    write_frame_stream: &mut FramedWrite<WriteHalf<TcpStream>, MqttCodec>,
 ) -> bool {
-    if connection_manager.tcp_connect_num_check() {
-        let packet_wrapper = MqttPacketWrapper {
-            protocol_version: MqttProtocol::Mqtt5.into(),
-            packet: response_packet_mqtt_distinct_by_reason(
-                &MqttProtocol::Mqtt5,
-                Some(DisconnectReasonCode::QuotaExceeded),
-            ),
-        };
-        match write_frame_stream.send(packet_wrapper).await {
-            Ok(_) => {}
-            Err(e) => error!("{}", e),
-        }
-
-        match write_frame_stream.close().await {
-            Ok(_) => {
-                error!(
-                    "tcp connection failed to establish from IP: {}",
-                    addr.to_string()
-                );
-            }
-            Err(e) => error!("{}", e),
-        }
-        return false;
+    if let Some(value) =
+        handle_tpc_connection_overflow(addr, connection_manager, write_frame_stream).await
+    {
+        return value;
     }
 
-    if is_connection_rate_exceeded() {
-        let packet_wrapper = MqttPacketWrapper {
-            protocol_version: MqttProtocol::Mqtt5.into(),
-            packet: response_packet_mqtt_distinct_by_reason(
-                &MqttProtocol::Mqtt5,
-                Some(DisconnectReasonCode::ConnectionRateExceeded),
-            ),
-        };
-        match write_frame_stream.send(packet_wrapper).await {
-            Ok(_) => {}
-            Err(e) => error!("{}", e),
-        }
-
-        match write_frame_stream.close().await {
-            Ok(_) => {
-                error!(
-                    "tcp connection failed to establish from IP: {}",
-                    addr.to_string()
-                );
-            }
-            Err(e) => error!("{}", e),
-        }
-        return false;
+    if let Some(value) = handle_connection_rate_exceeded(addr, write_frame_stream).await {
+        return value;
     }
     true
 }
@@ -107,10 +68,31 @@ pub async fn tcp_tls_establish_connection_check(
     addr: &SocketAddr,
     connection_manager: &Arc<ConnectionManager>,
     write_frame_stream: &mut FramedWrite<
-        tokio::io::WriteHalf<tokio_rustls::server::TlsStream<tokio::net::TcpStream>>,
+        WriteHalf<tokio_rustls::server::TlsStream<TcpStream>>,
         MqttCodec,
     >,
 ) -> bool {
+    if let Some(value) =
+        handle_tpc_connection_overflow(addr, connection_manager, write_frame_stream).await
+    {
+        return value;
+    }
+
+    if let Some(value) = handle_connection_rate_exceeded(addr, write_frame_stream).await {
+        return value;
+    }
+
+    true
+}
+
+async fn handle_tpc_connection_overflow<T>(
+    addr: &SocketAddr,
+    connection_manager: &Arc<ConnectionManager>,
+    write_frame_stream: &mut FramedWrite<WriteHalf<T>, MqttCodec>,
+) -> Option<bool>
+where
+    T: AsyncWriteExt + AsyncWrite,
+{
     if connection_manager.tcp_connect_num_check() {
         let packet_wrapper = MqttPacketWrapper {
             protocol_version: MqttProtocol::Mqtt5.into(),
@@ -133,9 +115,18 @@ pub async fn tcp_tls_establish_connection_check(
             }
             Err(e) => error!("{}", e),
         }
-        return false;
+        return Some(false);
     }
+    None
+}
 
+async fn handle_connection_rate_exceeded<T>(
+    addr: &SocketAddr,
+    write_frame_stream: &mut FramedWrite<WriteHalf<T>, MqttCodec>,
+) -> Option<bool>
+where
+    T: AsyncWriteExt + AsyncWrite,
+{
     if is_connection_rate_exceeded() {
         let packet_wrapper = MqttPacketWrapper {
             protocol_version: MqttProtocol::Mqtt5.into(),
@@ -158,9 +149,9 @@ pub async fn tcp_tls_establish_connection_check(
             }
             Err(e) => error!("{}", e),
         }
-        return false;
+        return Some(false);
     }
-    true
+    None
 }
 
 #[allow(clippy::too_many_arguments)]
diff --git a/src/mqtt-broker/src/lib.rs b/src/mqtt-broker/src/lib.rs
index 6fe1d234a..3b2945598 100644
--- a/src/mqtt-broker/src/lib.rs
+++ b/src/mqtt-broker/src/lib.rs
@@ -38,6 +38,7 @@ use storage::cluster::ClusterStorage;
 use storage_adapter::memory::MemoryStorageAdapter;
 // use storage_adapter::mysql::MySQLStorageAdapter;
 // use storage_adapter::rocksdb::RocksDBStorageAdapter;
+use crate::handler::flapping_detect::UpdateFlappingDetectCache;
 use storage_adapter::storage::StorageAdapter;
 use storage_adapter::StorageType;
 use subscribe::sub_exclusive::SubscribeExclusive;
@@ -157,6 +158,7 @@ where
         self.start_cluster_heartbeat_report(stop_send.clone());
         self.start_update_user_cache_thread(stop_send.clone());
         self.start_update_acl_cache_thread(stop_send.clone());
+        self.start_update_flapping_detect_cache_thread(stop_send.clone());
         self.start_push_server();
         self.start_system_topic_thread(stop_send.clone());
         self.awaiting_stop(stop_send);
@@ -319,6 +321,14 @@ where
         });
     }
 
+    fn start_update_flapping_detect_cache_thread(&self, stop_send: broadcast::Sender<bool>) {
+        let update_flapping_detect_cache =
+            UpdateFlappingDetectCache::new(stop_send, self.cache_manager.clone());
+        self.runtime.spawn(async move {
+            update_flapping_detect_cache.start_update().await;
+        });
+    }
+
     fn start_system_topic_thread(&self, stop_send: broadcast::Sender<bool>) {
         let cache_manager = self.cache_manager.clone();
         let message_storage_adapter = self.message_storage_adapter.clone();
diff --git a/src/mqtt-broker/src/observability/metrics/event_metrics.rs b/src/mqtt-broker/src/observability/metrics/event_metrics.rs
index 47e64d25c..77a75838a 100644
--- a/src/mqtt-broker/src/observability/metrics/event_metrics.rs
+++ b/src/mqtt-broker/src/observability/metrics/event_metrics.rs
@@ -87,5 +87,19 @@ mod tests {
             event_metrics::get_client_connection_counter("test_client_1".to_string()),
             1
         );
+
+        event_metrics::incr_client_connection_counter("test_client_1".to_string());
+
+        assert_eq!(
+            event_metrics::get_client_connection_counter("test_client_1".to_string()),
+            2
+        );
+
+        event_metrics::incr_client_connection_counter("test_client_2".to_string());
+
+        assert_eq!(
+            event_metrics::get_client_connection_counter("test_client_2".to_string()),
+            1
+        );
     }
 }
diff --git a/src/mqtt-broker/src/observability/slow/sub.rs b/src/mqtt-broker/src/observability/slow/sub.rs
index 9acb0a252..52c61923a 100644
--- a/src/mqtt-broker/src/observability/slow/sub.rs
+++ b/src/mqtt-broker/src/observability/slow/sub.rs
@@ -12,10 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::VecDeque;
-use std::fs::File;
-use std::path::PathBuf;
-
+use crate::handler::cache::CacheManager;
+use crate::handler::error::MqttBrokerError;
 use common_base::tools::{get_local_ip, now_second};
 use grep::matcher::Matcher;
 use grep::regex::RegexMatcher;
@@ -24,8 +22,10 @@ use grep::searcher::Searcher;
 use log::info;
 use protocol::broker_mqtt::broker_mqtt_admin::ListSlowSubscribeRequest;
 use serde::{Deserialize, Serialize};
-
-use crate::handler::error::MqttBrokerError;
+use std::collections::VecDeque;
+use std::fs::File;
+use std::path::PathBuf;
+use std::sync::Arc;
 
 #[derive(Debug, PartialEq, Serialize, Deserialize, Default, Clone)]
 pub struct SlowSubData {
@@ -114,6 +114,19 @@ pub fn read_slow_sub_record(
     Ok(matches_queue)
 }
 
+pub async fn enable_slow_sub(
+    cache_manager: &Arc<CacheManager>,
+    is_enable: bool,
+) -> Result<(), MqttBrokerError> {
+    let mut sub = cache_manager.get_slow_sub_config().clone();
+
+    sub.enable = is_enable;
+
+    cache_manager.set_slow_sub_config(sub).await?;
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/src/mqtt-broker/src/security/acl/metadata.rs b/src/mqtt-broker/src/security/acl/metadata.rs
index 611a1e46a..5a29871ec 100644
--- a/src/mqtt-broker/src/security/acl/metadata.rs
+++ b/src/mqtt-broker/src/security/acl/metadata.rs
@@ -12,9 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::handler::error::MqttBrokerError;
+use crate::handler::flapping_detect::FlappingDetectCondition;
+use common_base::enum_type::time_unit_enum::TimeUnit;
+use common_base::tools::{convert_seconds, now_second};
 use dashmap::DashMap;
 use metadata_struct::acl::mqtt_acl::{MqttAcl, MqttAclResourceType};
 use metadata_struct::acl::mqtt_blacklist::{MqttAclBlackList, MqttAclBlackListType};
+use metadata_struct::mqtt::cluster::MqttClusterDynamicFlappingDetect;
 
 #[derive(Clone)]
 pub struct AclMetadata {
@@ -29,6 +34,9 @@ pub struct AclMetadata {
     // acl
     pub acl_user: DashMap<String, Vec<MqttAcl>>,
     pub acl_client_id: DashMap<String, Vec<MqttAcl>>,
+
+    // connection jitter (client_id, FlappingDetectCondition)
+    pub flapping_detect_map: DashMap<String, FlappingDetectCondition>,
 }
 
 impl Default for AclMetadata {
@@ -49,9 +57,49 @@ impl AclMetadata {
 
             acl_user: DashMap::with_capacity(2),
             acl_client_id: DashMap::with_capacity(2),
+            flapping_detect_map: DashMap::new(),
         }
     }
 
+    pub fn get_flapping_detect_condition(
+        &self,
+        client_id: String,
+    ) -> Option<FlappingDetectCondition> {
+        if let Some(flapping_detect_condition) = self.flapping_detect_map.get(&client_id) {
+            return Some(flapping_detect_condition.clone());
+        }
+        None
+    }
+
+    pub fn add_flapping_detect_condition(
+        &self,
+        flapping_detect_condition: FlappingDetectCondition,
+    ) {
+        self.flapping_detect_map.insert(
+            flapping_detect_condition.client_id.clone(),
+            flapping_detect_condition,
+        );
+    }
+
+    pub fn remove_flapping_detect_condition(&self, client_id: &str) {
+        self.flapping_detect_map.remove(client_id);
+    }
+
+    pub async fn remove_flapping_detect_conditions(
+        &self,
+        config: MqttClusterDynamicFlappingDetect,
+    ) -> Result<(), MqttBrokerError> {
+        let current_time = now_second();
+        let window_time = convert_seconds(config.window_time as u64, TimeUnit::Minutes);
+        self.flapping_detect_map
+            .retain(|_, flapping_detect_condition| {
+                // we need retain elements within window_time,
+                // so now_seconds - first_request_time must less than window_time
+                current_time - flapping_detect_condition.first_request_time < window_time
+            });
+        Ok(())
+    }
+
     pub fn parse_mqtt_acl(&self, acl: MqttAcl) {
         match acl.resource_type {
             MqttAclResourceType::ClientId => {
@@ -190,13 +238,51 @@ impl AclMetadata {
 
 #[cfg(test)]
 mod test {
+    use crate::handler::flapping_detect::FlappingDetectCondition;
+    use crate::security::acl::metadata::AclMetadata;
     use common_base::tools::now_second;
     use metadata_struct::acl::mqtt_acl::{
         MqttAcl, MqttAclAction, MqttAclPermission, MqttAclResourceType,
     };
     use metadata_struct::acl::mqtt_blacklist::{MqttAclBlackList, MqttAclBlackListType};
+    use metadata_struct::mqtt::cluster::MqttClusterDynamicFlappingDetect;
 
-    use crate::security::acl::metadata::AclMetadata;
+    #[tokio::test]
+    pub async fn test_mqtt_remove_flapping_detect() {
+        let acl_metadata = AclMetadata::new();
+        let condition1 = FlappingDetectCondition {
+            client_id: "test_id_1".to_string(),
+            connect_times: 15,
+            first_request_time: now_second() - 10,
+        };
+        let condition2 = FlappingDetectCondition {
+            client_id: "test_id_2".to_string(),
+            connect_times: 15,
+            first_request_time: now_second() - 70,
+        };
+
+        acl_metadata.add_flapping_detect_condition(condition1);
+
+        acl_metadata.add_flapping_detect_condition(condition2);
+
+        assert!(acl_metadata.flapping_detect_map.contains_key("test_id_1"));
+        assert!(acl_metadata.flapping_detect_map.contains_key("test_id_2"));
+
+        let jitter_config = MqttClusterDynamicFlappingDetect {
+            enable: true,
+            window_time: 1,
+            max_client_connections: 15,
+            ban_time: 5,
+        };
+
+        acl_metadata
+            .remove_flapping_detect_conditions(jitter_config)
+            .await
+            .expect("TODO: panic message");
+
+        assert!(acl_metadata.flapping_detect_map.contains_key("test_id_1"));
+        assert!(!acl_metadata.flapping_detect_map.contains_key("test_id_2"));
+    }
 
     #[tokio::test]
     pub async fn parse_mqtt_acl_test() {
diff --git a/src/mqtt-broker/src/security/acl/mod.rs b/src/mqtt-broker/src/security/acl/mod.rs
index a3d960f4f..26afff0f3 100644
--- a/src/mqtt-broker/src/security/acl/mod.rs
+++ b/src/mqtt-broker/src/security/acl/mod.rs
@@ -29,7 +29,7 @@ use crate::handler::constant::WILDCARD_RESOURCE;
 pub mod metadata;
 
 pub fn is_allow_acl(
-    cache_mamanger: &Arc<CacheManager>,
+    cache_manager: &Arc<CacheManager>,
     connection: &MQTTConnection,
     topic_name: &str,
     action: MqttAclAction,
@@ -37,29 +37,22 @@ pub fn is_allow_acl(
     _: QoS,
 ) -> bool {
     // check super user
-    if is_super_user(cache_mamanger, &connection.login_user) {
+    if is_super_user(cache_manager, &connection.login_user) {
         return true;
     }
 
     // check blacklist
-    if is_blacklist(cache_mamanger, connection) {
+    if is_blacklist(cache_manager, connection) {
         return false;
     }
 
-    // chack acl
-    if is_acl_deny(cache_mamanger, connection, topic_name, action) {
+    // check acl
+    if is_acl_deny(cache_manager, connection, topic_name, action) {
         return false;
     }
 
     // check retain acl
-    if retain
-        && is_acl_deny(
-            cache_mamanger,
-            connection,
-            topic_name,
-            MqttAclAction::Retain,
-        )
-    {
+    if retain && is_acl_deny(cache_manager, connection, topic_name, MqttAclAction::Retain) {
         return false;
     }
 
@@ -76,7 +69,8 @@ fn is_super_user(cache_manager: &Arc<CacheManager>, username: &str) -> bool {
     false
 }
 
-fn is_blacklist(cache_manager: &Arc<CacheManager>, connection: &MQTTConnection) -> bool {
+pub fn is_blacklist(cache_manager: &Arc<CacheManager>, connection: &MQTTConnection) -> bool {
+    // todo: I believe this code can be refactored using the Chain of Responsibility pattern.
     // check user blacklist
     if let Some(data) = cache_manager
         .acl_metadata
diff --git a/src/mqtt-broker/src/security/mod.rs b/src/mqtt-broker/src/security/mod.rs
index 05e851910..34eb2cb37 100644
--- a/src/mqtt-broker/src/security/mod.rs
+++ b/src/mqtt-broker/src/security/mod.rs
@@ -36,6 +36,7 @@ use storage_adapter::StorageType;
 
 use crate::handler::cache::CacheManager;
 use crate::handler::error::MqttBrokerError;
+use crate::security::acl::is_blacklist;
 use crate::subscribe::sub_common::get_sub_topic_id_list;
 
 pub mod acl;
@@ -207,6 +208,11 @@ impl AuthDriver {
         Ok(())
     }
 
+    pub async fn allow_connect(&self, connection: &MQTTConnection) -> bool {
+        // check blacklist
+        is_blacklist(&self.cache_manager, connection)
+    }
+
     pub async fn allow_publish(
         &self,
         connection: &MQTTConnection,
@@ -236,7 +242,7 @@ impl AuthDriver {
                     &self.cache_manager,
                     connection,
                     &topic,
-                    MqttAclAction::Publish,
+                    MqttAclAction::Subscribe,
                     false,
                     filter.qos,
                 ) {
diff --git a/src/mqtt-broker/src/security/mysql/schema.rs b/src/mqtt-broker/src/security/mysql/schema.rs
index 38d50960f..3a6debde9 100644
--- a/src/mqtt-broker/src/security/mysql/schema.rs
+++ b/src/mqtt-broker/src/security/mysql/schema.rs
@@ -28,7 +28,7 @@ pub struct TAuthAcl {
     pub allow: u64,
     pub ipaddr: String,
     pub username: String,
-    pub clientid: String,
+    pub client_id: String,
     pub access: u64,
     pub topic: String,
 }
diff --git a/src/mqtt-broker/src/server/grpc/admin/services.rs b/src/mqtt-broker/src/server/grpc/admin/services.rs
index b47460969..f012e623d 100644
--- a/src/mqtt-broker/src/server/grpc/admin/services.rs
+++ b/src/mqtt-broker/src/server/grpc/admin/services.rs
@@ -29,16 +29,18 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
     CreateBlacklistReply, CreateBlacklistRequest, CreateTopicRewriteRuleReply,
     CreateTopicRewriteRuleRequest, CreateUserReply, CreateUserRequest, DeleteAclReply,
     DeleteAclRequest, DeleteBlacklistReply, DeleteBlacklistRequest, DeleteTopicRewriteRuleReply,
-    DeleteTopicRewriteRuleRequest, DeleteUserReply, DeleteUserRequest, EnableSlowSubScribeReply,
-    EnableSlowSubscribeRequest, ListAclReply, ListAclRequest, ListBlacklistReply,
-    ListBlacklistRequest, ListConnectionRaw, ListConnectionReply, ListConnectionRequest,
-    ListSlowSubScribeRaw, ListSlowSubscribeReply, ListSlowSubscribeRequest, ListTopicReply,
-    ListTopicRequest, ListUserReply, ListUserRequest, MqttTopic,
+    DeleteTopicRewriteRuleRequest, DeleteUserReply, DeleteUserRequest, EnableFlappingDetectReply,
+    EnableFlappingDetectRequest, EnableSlowSubScribeReply, EnableSlowSubscribeRequest,
+    ListAclReply, ListAclRequest, ListBlacklistReply, ListBlacklistRequest, ListConnectionRaw,
+    ListConnectionReply, ListConnectionRequest, ListSlowSubScribeRaw, ListSlowSubscribeReply,
+    ListSlowSubscribeRequest, ListTopicReply, ListTopicRequest, ListUserReply, ListUserRequest,
+    MqttTopic,
 };
 use tonic::{Request, Response, Status};
 
 use crate::handler::cache::CacheManager;
-use crate::observability::slow::sub::{read_slow_sub_record, SlowSubData};
+use crate::handler::flapping_detect::enable_flapping_detect;
+use crate::observability::slow::sub::{enable_slow_sub, read_slow_sub_record, SlowSubData};
 use crate::security::AuthDriver;
 use crate::server::connection_manager::ConnectionManager;
 use crate::storage::cluster::ClusterStorage;
@@ -257,6 +259,20 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         }
     }
 
+    async fn mqtt_broker_enable_flapping_detect(
+        &self,
+        request: Request<EnableFlappingDetectRequest>,
+    ) -> Result<Response<EnableFlappingDetectReply>, Status> {
+        let req = request.into_inner();
+
+        match enable_flapping_detect(&self.cache_manager, req).await {
+            Ok(_) => Ok(Response::new(EnableFlappingDetectReply {
+                is_enable: req.is_enable,
+            })),
+            Err(e) => Err(Status::cancelled(e.to_string())),
+        }
+    }
+
     // --- connection ---
     async fn mqtt_broker_list_connection(
         &self,
@@ -290,11 +306,7 @@ impl MqttBrokerAdminService for GrpcAdminServices {
     ) -> Result<Response<EnableSlowSubScribeReply>, Status> {
         let subscribe_request = request.into_inner();
 
-        match self
-            .cache_manager
-            .enable_slow_sub(subscribe_request.is_enable)
-            .await
-        {
+        match enable_slow_sub(&self.cache_manager, subscribe_request.is_enable).await {
             Ok(_) => Ok(Response::new(EnableSlowSubScribeReply {
                 is_enable: subscribe_request.is_enable,
             })),
diff --git a/src/mqtt-broker/src/server/http/server.rs b/src/mqtt-broker/src/server/http/server.rs
index 1129dfc5c..7d8116fa6 100644
--- a/src/mqtt-broker/src/server/http/server.rs
+++ b/src/mqtt-broker/src/server/http/server.rs
@@ -24,7 +24,7 @@ use super::connection::connection_list;
 use super::prometheus::metrics;
 use super::publish::http_publish;
 
-pub const ROUTE_PUBLISTH: &str = "/publish";
+pub const ROUTE_PUBLISH: &str = "/publish";
 pub const ROUTE_CONNECTION: &str = "/connection";
 pub const ROUTE_METRICS: &str = "/metrics";
 
@@ -52,7 +52,7 @@ pub async fn start_http_server(state: HttpServerState) -> Result<(), CommonError
 
 fn routes_v1(state: HttpServerState) -> Router {
     let meta = Router::new()
-        .route(ROUTE_PUBLISTH, get(http_publish))
+        .route(ROUTE_PUBLISH, get(http_publish))
         .route(ROUTE_CONNECTION, get(connection_list))
         .route(ROUTE_METRICS, get(metrics));
 
diff --git a/src/mqtt-broker/src/server/tcp/tcp_server.rs b/src/mqtt-broker/src/server/tcp/tcp_server.rs
index 5d3d5f911..0216678ff 100644
--- a/src/mqtt-broker/src/server/tcp/tcp_server.rs
+++ b/src/mqtt-broker/src/server/tcp/tcp_server.rs
@@ -92,7 +92,7 @@ pub(crate) async fn acceptor_process(
 
                                 let (connection_stop_sx, connection_stop_rx) = mpsc::channel::<bool>(1);
                                 let connection = NetworkConnection::new(
-                                    crate::server::connection::NetworkConnectionType::Tcp,
+                                    NetworkConnectionType::Tcp,
                                     addr,
                                     Some(connection_stop_sx.clone())
                                 );
@@ -113,7 +113,7 @@ pub(crate) async fn acceptor_process(
 }
 
 fn read_frame_process(
-    mut read_frame_stream: FramedRead<tokio::io::ReadHalf<tokio::net::TcpStream>, MqttCodec>,
+    mut read_frame_stream: FramedRead<io::ReadHalf<tokio::net::TcpStream>, MqttCodec>,
     connection: NetworkConnection,
     request_queue_sx: Sender<RequestPackage>,
     mut connection_stop_rx: Receiver<bool>,
diff --git a/src/mqtt-broker/src/subscribe/sub_exclusive.rs b/src/mqtt-broker/src/subscribe/sub_exclusive.rs
index 06002229d..252acf93e 100644
--- a/src/mqtt-broker/src/subscribe/sub_exclusive.rs
+++ b/src/mqtt-broker/src/subscribe/sub_exclusive.rs
@@ -217,7 +217,7 @@ where
         let record_offset = record.offset.unwrap();
 
         // build publish params
-        let sub_pub_param = if let Some(params) = buile_pub_message(
+        let sub_pub_param = if let Some(params) = build_pub_message(
             record.to_owned(),
             group_id,
             qos,
@@ -306,7 +306,7 @@ where
     Ok(Some(results.last().unwrap().offset.unwrap()))
 }
 
-async fn buile_pub_message(
+async fn build_pub_message(
     record: Record,
     group_id: &str,
     qos: &QoS,
diff --git a/src/protocol/src/broker_mqtt/proto/admin.proto b/src/protocol/src/broker_mqtt/proto/admin.proto
index 2143c4931..cd97d5d60 100644
--- a/src/protocol/src/broker_mqtt/proto/admin.proto
+++ b/src/protocol/src/broker_mqtt/proto/admin.proto
@@ -25,6 +25,9 @@ service MQTTBrokerAdminService {
 
     rpc mqtt_broker_create_blacklist(CreateBlacklistRequest) returns(CreateBlacklistReply) {}
 
+    // flapping detect
+    rpc mqtt_broker_enable_flapping_detect(EnableFlappingDetectRequest) returns(EnableFlappingDetectReply) {}
+
     // connection
     rpc mqtt_broker_list_connection(ListConnectionRequest) returns(ListConnectionReply){}
 
@@ -33,6 +36,8 @@ service MQTTBrokerAdminService {
     rpc mqtt_broker_list_slow_subscribe(ListSlowSubscribeRequest) returns(ListSlowSubscribeReply){}
     rpc mqtt_broker_list_topic(ListTopicRequest) returns(ListTopicReply){}
 
+
+
     // topic rewrite rule
     rpc mqtt_broker_delete_topic_rewrite_rule(DeleteTopicRewriteRuleRequest) returns(DeleteTopicRewriteRuleReply) {}
     rpc mqtt_broker_create_topic_rewrite_rule(CreateTopicRewriteRuleRequest) returns(CreateTopicRewriteRuleReply) {}
@@ -151,6 +156,18 @@ message ListConnectionRaw {
     string info = 5;
 }
 
+// flapping detect
+message EnableFlappingDetectRequest {
+    bool is_enable = 1;
+    uint32 window_time = 2;
+    uint32 max_client_connections = 3;
+    uint32 ban_time = 4;
+}
+
+message EnableFlappingDetectReply {
+    bool is_enable = 1;
+}
+
 // --------- observability service --------
 // --------- slow subscribe feature -------
 message EnableSlowSubscribeRequest {


Test patch:
diff --git a/tests/tests/mqtt_client/flapping_detect_test.rs b/tests/tests/mqtt_client/flapping_detect_test.rs
new file mode 100644
index 000000000..4f1d6404d
--- /dev/null
+++ b/tests/tests/mqtt_client/flapping_detect_test.rs
@@ -0,0 +1,86 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests {
+    use common_base::error::common::CommonError;
+    use grpc_clients::mqtt::admin::call::mqtt_broker_enable_flapping_detect;
+    use grpc_clients::pool::ClientPool;
+
+    use crate::mqtt_protocol::common::broker_grpc_addr;
+    use protocol::broker_mqtt::broker_mqtt_admin::{
+        EnableFlappingDetectReply, EnableFlappingDetectRequest,
+    };
+    use std::sync::Arc;
+
+    async fn open_flapping_detect(
+        request: EnableFlappingDetectRequest,
+    ) -> Result<EnableFlappingDetectReply, CommonError> {
+        let client_pool = Arc::new(ClientPool::new(3));
+        let grpc_addr = vec![broker_grpc_addr()];
+        mqtt_broker_enable_flapping_detect(&client_pool, &grpc_addr, request).await
+    }
+
+    #[tokio::test]
+    async fn test_enable_flapping_detect() {
+        // per
+        let request = EnableFlappingDetectRequest {
+            is_enable: false,
+            window_time: 0,
+            max_client_connections: 0,
+            ban_time: 0,
+        };
+
+        // result
+        let reply = EnableFlappingDetectReply { is_enable: false };
+
+        // action
+        match open_flapping_detect(request).await {
+            Ok(data) => {
+                assert_eq!(data, reply);
+            }
+
+            Err(e) => {
+                eprintln!("Failed enable_slow_subscribe: {:?}", e);
+                std::process::exit(1);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_enable_flapping_detect_is_true() {
+        // per
+        let request = EnableFlappingDetectRequest {
+            is_enable: true,
+            window_time: 1,
+            max_client_connections: 15,
+            ban_time: 1,
+        };
+
+        // result
+        let reply = EnableFlappingDetectReply { is_enable: true };
+
+        // action
+        match open_flapping_detect(request).await {
+            Ok(data) => {
+                assert_eq!(data, reply);
+            }
+
+            Err(e) => {
+                eprintln!("Failed enable_slow_subscribe: {:?}", e);
+                std::process::exit(1);
+            }
+        }
+    }
+}
diff --git a/tests/tests/mqtt_client/mod.rs b/tests/tests/mqtt_client/mod.rs
index 02c2020f0..f11b92c54 100644
--- a/tests/tests/mqtt_client/mod.rs
+++ b/tests/tests/mqtt_client/mod.rs
@@ -12,4 +12,5 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod flapping_detect_test;
 mod slow_sub_test;
diff --git a/tests/tests/mqtt_client/slow_sub_test.rs b/tests/tests/mqtt_client/slow_sub_test.rs
index 2fb59ef2c..33a3563b4 100644
--- a/tests/tests/mqtt_client/slow_sub_test.rs
+++ b/tests/tests/mqtt_client/slow_sub_test.rs
@@ -21,7 +21,7 @@ mod tests {
     };
     use grpc_clients::pool::ClientPool;
     use protocol::broker_mqtt::broker_mqtt_admin::{
-        EnableSlowSubscribeRequest, ListSlowSubscribeRequest,
+        EnableSlowSubScribeReply, EnableSlowSubscribeRequest, ListSlowSubscribeRequest,
     };
 
     use crate::mqtt_protocol::common::broker_grpc_addr;
@@ -33,9 +33,11 @@ mod tests {
 
         let request = EnableSlowSubscribeRequest { is_enable: true };
 
+        let reply = EnableSlowSubScribeReply { is_enable: true };
+
         match mqtt_broker_enable_slow_subscribe(&client_pool, &grpc_addr, request).await {
             Ok(data) => {
-                println!("{:?}", data);
+                assert_eq!(reply, data);
             }
 
             Err(e) => {
diff --git a/tests/tests/mqtt_protocol/common.rs b/tests/tests/mqtt_protocol/common.rs
index 57ed836a1..e97473366 100644
--- a/tests/tests/mqtt_protocol/common.rs
+++ b/tests/tests/mqtt_protocol/common.rs
@@ -15,6 +15,7 @@
 use std::process;
 use std::time::Duration;
 
+use crate::mqtt_protocol::connect_suite::ClientTestProperties;
 use paho_mqtt::{
     Client, ConnectOptions, ConnectOptionsBuilder, CreateOptions, CreateOptionsBuilder,
     DisconnectOptionsBuilder, Properties, PropertyCode, ReasonCode, SslOptionsBuilder,
@@ -152,6 +153,42 @@ pub fn build_v5_conn_pros_by_user_information(
         .finalize()
 }
 
+#[allow(dead_code)]
+pub fn build_conn_pros(
+    client_test_properties: ClientTestProperties,
+    err_pwd: bool,
+) -> ConnectOptions {
+    let pwd = if err_pwd { err_password() } else { password() };
+    let mut conn_opts = if client_test_properties.ws {
+        ConnectOptionsBuilder::new_ws()
+    } else {
+        ConnectOptionsBuilder::with_mqtt_version(client_test_properties.mqtt_version)
+    };
+    if client_test_properties.ssl {
+        let ssl_opts = SslOptionsBuilder::new()
+            .trust_store(get_cargo_manifest_dir())
+            .unwrap()
+            .verify(false)
+            .disable_default_trust_store(false)
+            .finalize();
+        conn_opts.ssl_options(ssl_opts);
+    }
+    conn_opts
+        .keep_alive_interval(Duration::from_secs(600))
+        .clean_session(true)
+        .connect_timeout(Duration::from_secs(50))
+        .user_name(username())
+        .password(pwd)
+        .finalize()
+}
+
+fn get_cargo_manifest_dir() -> String {
+    format!(
+        "{}/../config/example/certs/ca.pem",
+        env!("CARGO_MANIFEST_DIR")
+    )
+}
+
 #[allow(dead_code)]
 pub fn build_v3_conn_pros(mqtt_version: u32, err_pwd: bool, ws: bool, ssl: bool) -> ConnectOptions {
     let pwd = if err_pwd { err_password() } else { password() };
diff --git a/tests/tests/mqtt_protocol/connect34_test.rs b/tests/tests/mqtt_protocol/connect34_test.rs
index 96821a57a..4c42281e6 100644
--- a/tests/tests/mqtt_protocol/connect34_test.rs
+++ b/tests/tests/mqtt_protocol/connect34_test.rs
@@ -14,134 +14,211 @@
 
 #[cfg(test)]
 mod tests {
+    use crate::mqtt_protocol::common::{
+        broker_addr, broker_grpc_addr, broker_ssl_addr, broker_ws_addr, broker_wss_addr,
+        build_conn_pros, build_create_pros, distinct_conn,
+    };
+    use crate::mqtt_protocol::connect_suite::ClientTestProperties;
+    use crate::mqtt_protocol::connect_suite::{session_present_test, wrong_password_test};
+    use common_base::tools::unique_id;
+    use grpc_clients::mqtt::admin::call::mqtt_broker_enable_flapping_detect;
+    use grpc_clients::pool::ClientPool;
+    use paho_mqtt::Client;
+    use protocol::broker_mqtt::broker_mqtt_admin::EnableFlappingDetectRequest;
     use std::process;
+    use std::sync::Arc;
 
-    use common_base::tools::unique_id;
-    use paho_mqtt::{Client, ReasonCode};
+    #[tokio::test]
+    async fn client_connect_wrong_password_test_v3() {
+        let client_v3_properties = ClientTestProperties {
+            mqtt_version: 3,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        // connect_test
+        wrong_password_test(client_v3_properties.clone());
+        session_present_test(client_v3_properties.clone());
+    }
 
-    use crate::mqtt_protocol::common::{
-        broker_addr, broker_ssl_addr, broker_ws_addr, broker_wss_addr, build_create_pros,
-        build_v3_conn_pros, distinct_conn,
-    };
+    #[tokio::test]
+    async fn client_connect_session_present_test_v3() {
+        let client_v3_properties = ClientTestProperties {
+            mqtt_version: 3,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        session_present_test(client_v3_properties.clone());
+    }
 
     #[tokio::test]
-    async fn client34_connect_test() {
-        let mqtt_version = 3;
-        let client_id = unique_id();
-        let addr = broker_addr();
-        v3_wrong_password_test(mqtt_version, &client_id, &addr, false, false);
-        v3_session_present_test(mqtt_version, &client_id, &addr, false, false);
-
-        let mqtt_version = 4;
-        let client_id = unique_id();
-        let addr = broker_addr();
-        v3_wrong_password_test(mqtt_version, &client_id, &addr, false, false);
-        v3_session_present_test(mqtt_version, &client_id, &addr, false, false);
+    async fn client_connect_ssl_test_v3() {
+        let mut client_v3_properties = ClientTestProperties {
+            mqtt_version: 3,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        client_v3_properties.addr = broker_ssl_addr();
+        client_v3_properties.ssl = true;
+
+        wrong_password_test(client_v3_properties.clone());
+        session_present_test(client_v3_properties.clone());
     }
 
     #[tokio::test]
-    async fn client34_connect_ssl_test() {
-        let mqtt_version = 3;
-        let client_id = unique_id();
-        let addr = broker_ssl_addr();
-        v3_wrong_password_test(mqtt_version, &client_id, &addr, false, true);
-        v3_session_present_test(mqtt_version, &client_id, &addr, false, true);
-
-        let mqtt_version = 4;
-        let client_id = unique_id();
-        let addr = broker_ssl_addr();
-        v3_wrong_password_test(mqtt_version, &client_id, &addr, false, true);
-        v3_session_present_test(mqtt_version, &client_id, &addr, false, true);
+    async fn client_connect_test_v4() {
+        let client_properties_v4 = ClientTestProperties {
+            mqtt_version: 4,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        // connect_test
+        wrong_password_test(client_properties_v4.clone());
+        session_present_test(client_properties_v4.clone());
     }
 
     #[tokio::test]
-    async fn client4_connect_ws_test() {
-        let mqtt_version = 4;
-        let client_id = unique_id();
-        let addr = broker_ws_addr();
+    async fn client_connect_ssl_test_v4() {
+        let mut client_properties_v4 = ClientTestProperties {
+            mqtt_version: 4,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        client_properties_v4.addr = broker_ssl_addr();
+        client_properties_v4.ssl = true;
+
+        wrong_password_test(client_properties_v4.clone());
+        session_present_test(client_properties_v4.clone());
+    }
 
-        v3_wrong_password_test(mqtt_version, &client_id, &addr, true, false);
-        v3_session_present_test(mqtt_version, &client_id, &addr, true, false);
+    #[tokio::test]
+    async fn client_connect_ws_test_v4() {
+        let mut client_properties_v4 = ClientTestProperties {
+            mqtt_version: 4,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        // ws_test
+        client_properties_v4.addr = broker_ws_addr();
+        client_properties_v4.ws = true;
+        client_properties_v4.ssl = false;
+
+        wrong_password_test(client_properties_v4.clone());
+        session_present_test(client_properties_v4.clone());
     }
 
     #[tokio::test]
-    async fn client4_connect_wss_test() {
-        let mqtt_version = 4;
-        let client_id = unique_id();
-        let addr = broker_wss_addr();
-        v3_wrong_password_test(mqtt_version, &client_id, &addr, true, true);
-        v3_session_present_test(mqtt_version, &client_id, &addr, true, true);
+    async fn client_connect_wss_test_v4() {
+        let mut client_properties_v4 = ClientTestProperties {
+            mqtt_version: 4,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+        // ws_test
+        client_properties_v4.addr = broker_wss_addr();
+        client_properties_v4.ws = true;
+        client_properties_v4.ssl = true;
+
+        wrong_password_test(client_properties_v4.clone());
+        session_present_test(client_properties_v4.clone());
+    }
+
+    async fn open_flapping_detect() {
+        let client_pool = Arc::new(ClientPool::new(3));
+        let grpc_addr = vec![broker_grpc_addr()];
+
+        let request = EnableFlappingDetectRequest {
+            is_enable: true,
+            window_time: 20,
+            max_client_connections: 2,
+            ban_time: 5,
+        };
+
+        let _reply = mqtt_broker_enable_flapping_detect(&client_pool, &grpc_addr, request)
+            .await
+            .unwrap();
     }
 
-    fn v3_wrong_password_test(mqtt_version: u32, client_id: &str, addr: &str, ws: bool, ssl: bool) {
-        let create_opts = build_create_pros(client_id, addr);
+    #[tokio::test]
+    async fn client_flapping_detect_test() {
+        open_flapping_detect().await;
+
+        let client_test_properties = ClientTestProperties {
+            mqtt_version: 3,
+            client_id: unique_id(),
+            addr: broker_addr(),
+            ws: false,
+            ssl: false,
+        };
+
+        let create_opts = build_create_pros(
+            &client_test_properties.client_id,
+            &client_test_properties.addr,
+        );
         let cli = Client::new(create_opts).unwrap_or_else(|err| {
             println!("Error creating the client: {:?}", err);
             process::exit(1);
         });
 
-        let conn_opts = build_v3_conn_pros(mqtt_version, true, ws, ssl);
-        println!("{:?}", conn_opts);
-        let err = cli.connect(conn_opts).unwrap_err();
-        println!("Unable to connect:\n\t{:?}", err);
-    }
+        let conn_opts = build_conn_pros(client_test_properties.clone(), false);
+
+        assert!(cli.connect(conn_opts.clone()).is_ok());
 
-    fn v3_session_present_test(
-        mqtt_version: u32,
-        client_id: &str,
-        addr: &str,
-        ws: bool,
-        ssl: bool,
-    ) {
-        let create_opts = build_create_pros(client_id, addr);
-        let cli = Client::new(create_opts).unwrap();
-        let conn_opts = build_v3_conn_pros(mqtt_version, false, ws, ssl);
-        let response = cli.connect(conn_opts).unwrap();
-        let resp = response.connect_response().unwrap();
-        if ws {
-            if ssl {
-                assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
-            } else {
-                assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
-            }
-            assert_eq!(4, resp.mqtt_version);
-        } else {
-            if ssl {
-                assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
-            } else {
-                assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
-            }
-            assert_eq!(mqtt_version, resp.mqtt_version);
-        }
-        assert!(resp.session_present);
-        assert_eq!(response.reason_code(), ReasonCode::Success);
         distinct_conn(cli);
 
-        let create_opts = build_create_pros(client_id, addr);
-
-        let cli = Client::new(create_opts).unwrap();
-
-        let conn_opts = build_v3_conn_pros(mqtt_version, false, ws, ssl);
-
-        let response = cli.connect(conn_opts).unwrap();
-        let resp = response.connect_response().unwrap();
-        if ws {
-            if ssl {
-                assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
-            } else {
-                assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
-            }
-            assert_eq!(4, resp.mqtt_version);
-        } else {
-            if ssl {
-                assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
-            } else {
-                assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
-            }
-            assert_eq!(mqtt_version, resp.mqtt_version);
-        }
-        assert!(!resp.session_present);
-        assert_eq!(response.reason_code(), ReasonCode::Success);
+        let create_opts = build_create_pros(
+            &client_test_properties.client_id,
+            &client_test_properties.addr,
+        );
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_conn_pros(client_test_properties.clone(), false);
+
+        assert!(cli.connect(conn_opts.clone()).is_ok());
+
+        distinct_conn(cli);
+
+        let create_opts = build_create_pros(
+            &client_test_properties.client_id,
+            &client_test_properties.addr,
+        );
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_conn_pros(client_test_properties.clone(), false);
+
+        assert!(cli.connect(conn_opts.clone()).is_ok());
+
         distinct_conn(cli);
+
+        let create_opts = build_create_pros(
+            &client_test_properties.client_id,
+            &client_test_properties.addr,
+        );
+        let cli = Client::new(create_opts).unwrap_or_else(|err| {
+            println!("Error creating the client: {:?}", err);
+            process::exit(1);
+        });
+
+        let conn_opts = build_conn_pros(client_test_properties.clone(), false);
+        assert!(cli.connect(conn_opts.clone()).is_err());
     }
 }
diff --git a/tests/tests/mqtt_protocol/connect_suite.rs b/tests/tests/mqtt_protocol/connect_suite.rs
new file mode 100644
index 000000000..af4fb2cb7
--- /dev/null
+++ b/tests/tests/mqtt_protocol/connect_suite.rs
@@ -0,0 +1,89 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::mqtt_protocol::common::{
+    broker_addr, broker_ssl_addr, broker_ws_addr, broker_wss_addr, build_conn_pros,
+    build_create_pros, distinct_conn,
+};
+use paho_mqtt::{Client, ReasonCode};
+use std::process;
+
+#[derive(Debug, Clone)]
+pub struct ClientTestProperties {
+    pub(crate) mqtt_version: u32,
+    pub(crate) client_id: String,
+    pub(crate) addr: String,
+    pub(crate) ws: bool,
+    pub(crate) ssl: bool,
+}
+
+pub fn wrong_password_test(client_test_properties: ClientTestProperties) {
+    let create_opts = build_create_pros(
+        &client_test_properties.client_id,
+        &client_test_properties.addr,
+    );
+    let cli = Client::new(create_opts).unwrap_or_else(|err| {
+        println!("Error creating the client: {:?}", err);
+        process::exit(1);
+    });
+
+    let conn_opts = build_conn_pros(client_test_properties.clone(), true);
+
+    println!("{:?}", conn_opts);
+    let err = cli.connect(conn_opts).unwrap_err();
+    println!("Unable to connect:\n\t{:?}", err);
+}
+
+pub fn session_present_test(client_test_properties: ClientTestProperties) {
+    create_session_connection(client_test_properties.clone(), true);
+
+    create_session_connection(client_test_properties.clone(), false);
+}
+
+fn create_session_connection(client_test_properties: ClientTestProperties, present: bool) {
+    let create_opts = build_create_pros(
+        &client_test_properties.client_id,
+        &client_test_properties.addr,
+    );
+    let cli = Client::new(create_opts).unwrap();
+
+    let conn_opts = build_conn_pros(client_test_properties.clone(), false);
+    let response = cli.connect(conn_opts).unwrap();
+
+    let resp = response.connect_response().unwrap();
+    if client_test_properties.ws {
+        if client_test_properties.ssl {
+            assert_eq!(format!("wss://{}", resp.server_uri), broker_wss_addr());
+        } else {
+            assert_eq!(format!("ws://{}", resp.server_uri), broker_ws_addr());
+        }
+        assert_eq!(4, resp.mqtt_version);
+    } else {
+        if client_test_properties.ssl {
+            assert_eq!(format!("mqtts://{}", resp.server_uri), broker_ssl_addr());
+        } else {
+            assert_eq!(format!("tcp://{}", resp.server_uri), broker_addr());
+        }
+        assert_eq!(client_test_properties.mqtt_version, resp.mqtt_version);
+    }
+    if present {
+        assert!(resp.session_present);
+    } else {
+        assert!(!resp.session_present);
+    }
+
+    assert_eq!(response.reason_code(), ReasonCode::Success);
+
+    distinct_conn(cli);
+}
diff --git a/tests/tests/mqtt_protocol/mod.rs b/tests/tests/mqtt_protocol/mod.rs
index db1775f5d..d7498d594 100644
--- a/tests/tests/mqtt_protocol/mod.rs
+++ b/tests/tests/mqtt_protocol/mod.rs
@@ -17,6 +17,7 @@ pub mod common;
 pub mod connect34_test;
 pub mod connect5_test;
 pub mod connect_packet_size_test;
+mod connect_suite;
 pub mod keep_alive_test;
 pub mod lastwill_message_test;
 pub mod permission34_test;
diff --git a/tests/tests/mqtt_protocol/req_resp_test.rs b/tests/tests/mqtt_protocol/req_resp_test.rs
index 1769ffdaa..b97c96891 100644
--- a/tests/tests/mqtt_protocol/req_resp_test.rs
+++ b/tests/tests/mqtt_protocol/req_resp_test.rs
@@ -22,7 +22,7 @@ mod tests {
     };
 
     #[tokio::test]
-    async fn client5_reuqset_response_test() {
+    async fn client5_request_response_test() {
         let sub_qos = &[0, 0];
 
         let topic = unique_id();