diff --git a/apisix/plugins/clickhouse-logger.lua b/apisix/plugins/clickhouse-logger.lua
index 793a8d462ec9..b3e791f079b1 100644
--- a/apisix/plugins/clickhouse-logger.lua
+++ b/apisix/plugins/clickhouse-logger.lua
@@ -17,6 +17,7 @@
 
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 local log_util        = require("apisix.utils.log-util")
+local plugin          = require("apisix.plugin")
 local core            = require("apisix.core")
 local http            = require("resty.http")
 local url             = require("net.url")
@@ -71,7 +72,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -174,9 +180,12 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = log_util.get_log_entry(plugin_name, conf, ctx)
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -201,7 +210,7 @@ function _M.log(conf, ctx)
         return send_http_data(conf, data)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/elasticsearch-logger.lua b/apisix/plugins/elasticsearch-logger.lua
index 09dcbd795c85..3d9da16bb62e 100644
--- a/apisix/plugins/elasticsearch-logger.lua
+++ b/apisix/plugins/elasticsearch-logger.lua
@@ -19,7 +19,7 @@ local core            = require("apisix.core")
 local http            = require("resty.http")
 local log_util        = require("apisix.utils.log-util")
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
-
+local plugin          = require("apisix.plugin")
 local ngx             = ngx
 local str_format      = core.string.format
 local math_random     = math.random
@@ -104,7 +104,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -264,9 +269,12 @@ function _M.access(conf)
 end
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = get_logger_entry(conf, ctx)
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -274,7 +282,8 @@ function _M.log(conf, ctx)
         return send_to_elasticsearch(conf, entries)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, process)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
+                                                       process, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/google-cloud-logging.lua b/apisix/plugins/google-cloud-logging.lua
index 62ca991c0e80..09d6eb5f907f 100644
--- a/apisix/plugins/google-cloud-logging.lua
+++ b/apisix/plugins/google-cloud-logging.lua
@@ -16,6 +16,7 @@
 --
 
 local core            = require("apisix.core")
+local plugin          = require("apisix.plugin")
 local tostring        = tostring
 local http            = require("resty.http")
 local log_util        = require("apisix.utils.log-util")
@@ -110,7 +111,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -241,6 +247,9 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local oauth, err = core.lrucache.plugin_ctx(lrucache, ctx, nil,
                                                 create_oauth_object, conf)
     if not oauth then
@@ -250,7 +259,7 @@ function _M.log(conf, ctx)
 
     local entry = get_logger_entry(conf, ctx, oauth)
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -258,7 +267,8 @@ function _M.log(conf, ctx)
         return send_to_google(oauth, entries)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, process)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
+                                                       process, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/http-logger.lua b/apisix/plugins/http-logger.lua
index 44f84acc907d..d3259e5db53f 100644
--- a/apisix/plugins/http-logger.lua
+++ b/apisix/plugins/http-logger.lua
@@ -16,6 +16,7 @@
 --
 
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
+local plugin          = require("apisix.plugin")
 local log_util        = require("apisix.utils.log-util")
 local core            = require("apisix.core")
 local http            = require("resty.http")
@@ -63,7 +64,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -168,13 +174,16 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = log_util.get_log_entry(plugin_name, conf, ctx)
 
     if not entry.route_id then
         entry.route_id = "no-matched"
     end
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -216,7 +225,7 @@ function _M.log(conf, ctx)
         return send_http_data(conf, data)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/loki-logger.lua b/apisix/plugins/loki-logger.lua
index 6ff5311ebcaa..bb7c614f17f0 100644
--- a/apisix/plugins/loki-logger.lua
+++ b/apisix/plugins/loki-logger.lua
@@ -18,6 +18,7 @@
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
 local log_util        = require("apisix.utils.log-util")
 local core            = require("apisix.core")
+local plugin          = require("apisix.plugin")
 local http            = require("resty.http")
 local new_tab         = require("table.new")
 
@@ -115,7 +116,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -193,6 +199,9 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = log_util.get_log_entry(plugin_name, conf, ctx)
 
     if not entry.route_id then
@@ -205,7 +214,7 @@ function _M.log(conf, ctx)
     -- and then add 6 zeros by string concatenation
     entry.loki_log_time = tostring(ngx.req.start_time() * 1000) .. "000000"
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -244,7 +253,7 @@ function _M.log(conf, ctx)
         return send_http_data(conf, data)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/rocketmq-logger.lua b/apisix/plugins/rocketmq-logger.lua
index 2f0cd5b4ba59..9dd5b2a93ec2 100644
--- a/apisix/plugins/rocketmq-logger.lua
+++ b/apisix/plugins/rocketmq-logger.lua
@@ -15,6 +15,7 @@
 -- limitations under the License.
 --
 local core     = require("apisix.core")
+local plugin   = require("apisix.plugin")
 local log_util = require("apisix.utils.log-util")
 local producer = require ("resty.rocketmq.producer")
 local acl_rpchook = require("resty.rocketmq.acl_rpchook")
@@ -77,7 +78,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -138,6 +144,9 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry
     if conf.meta_format == "origin" then
         entry = log_util.get_req_original(ctx, conf)
@@ -145,7 +154,7 @@ function _M.log(conf, ctx)
         entry = log_util.get_log_entry(plugin_name, conf, ctx)
     end
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -184,7 +193,7 @@ function _M.log(conf, ctx)
         return send_rocketmq_data(conf, data, prod)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/skywalking-logger.lua b/apisix/plugins/skywalking-logger.lua
index 8a7e309cdabb..c7c8dcae8b14 100644
--- a/apisix/plugins/skywalking-logger.lua
+++ b/apisix/plugins/skywalking-logger.lua
@@ -16,6 +16,7 @@
 --
 
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
+local plugin          = require("apisix.plugin")
 local log_util        = require("apisix.utils.log-util")
 local core            = require("apisix.core")
 local http            = require("resty.http")
@@ -64,7 +65,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -139,6 +145,9 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local log_body = log_util.get_log_entry(plugin_name, conf, ctx)
     local trace_context
     local sw_header = ngx.req.get_headers()["sw8"]
@@ -173,7 +182,7 @@ function _M.log(conf, ctx)
         endpoint = ctx.var.uri,
     }
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -187,7 +196,7 @@ function _M.log(conf, ctx)
         return send_http_data(conf, data)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/splunk-hec-logging.lua b/apisix/plugins/splunk-hec-logging.lua
index c93b2736eeea..2024cc9fea2c 100644
--- a/apisix/plugins/splunk-hec-logging.lua
+++ b/apisix/plugins/splunk-hec-logging.lua
@@ -21,6 +21,7 @@ local ngx_now         = ngx.now
 local http            = require("resty.http")
 local log_util        = require("apisix.utils.log-util")
 local bp_manager_mod  = require("apisix.utils.batch-processor-manager")
+local plugin          = require("apisix.plugin")
 local table_insert    = core.table.insert
 local table_concat    = core.table.concat
 local ipairs          = ipairs
@@ -75,6 +76,11 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
         }
     },
 }
@@ -169,9 +175,12 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = get_logger_entry(conf, ctx)
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -179,7 +188,8 @@ function _M.log(conf, ctx)
         return send_to_splunk(conf, entries)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, process)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
+                                                        process, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/tcp-logger.lua b/apisix/plugins/tcp-logger.lua
index 7482fe515dee..9b3df668ea64 100644
--- a/apisix/plugins/tcp-logger.lua
+++ b/apisix/plugins/tcp-logger.lua
@@ -16,6 +16,7 @@
 --
 local core     = require("apisix.core")
 local log_util = require("apisix.utils.log-util")
+local plugin   = require("apisix.plugin")
 local bp_manager_mod = require("apisix.utils.batch-processor-manager")
 local plugin_name = "tcp-logger"
 local tostring = tostring
@@ -58,7 +59,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -132,9 +138,12 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = log_util.get_log_entry(plugin_name, conf, ctx)
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -154,7 +163,7 @@ function _M.log(conf, ctx)
         return send_tcp_data(conf, data)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/tencent-cloud-cls.lua b/apisix/plugins/tencent-cloud-cls.lua
index 38fe56503202..f8dac160ed87 100644
--- a/apisix/plugins/tencent-cloud-cls.lua
+++ b/apisix/plugins/tencent-cloud-cls.lua
@@ -16,6 +16,7 @@
 --
 
 local core = require("apisix.core")
+local plugin = require("apisix.plugin")
 local log_util = require("apisix.utils.log-util")
 local bp_manager_mod = require("apisix.utils.batch-processor-manager")
 local cls_sdk = require("apisix.plugins.tencent-cloud-cls.cls-sdk")
@@ -67,7 +68,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -112,6 +118,9 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     -- sample if set
     if not ctx.cls_sample then
         core.log.debug("cls not sampled, skip log")
@@ -126,7 +135,7 @@ function _M.log(conf, ctx)
         end
     end
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -139,7 +148,8 @@ function _M.log(conf, ctx)
         return sdk:send_to_cls(entries)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, process)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx,
+                                                        process, max_pending_entries)
 end
 
 
diff --git a/apisix/plugins/udp-logger.lua b/apisix/plugins/udp-logger.lua
index 75e8bba31740..a2333b0b5b2e 100644
--- a/apisix/plugins/udp-logger.lua
+++ b/apisix/plugins/udp-logger.lua
@@ -16,6 +16,7 @@
 --
 local core = require("apisix.core")
 local log_util = require("apisix.utils.log-util")
+local plugin = require("apisix.plugin")
 local bp_manager_mod = require("apisix.utils.batch-processor-manager")
 local plugin_name = "udp-logger"
 local tostring = tostring
@@ -56,7 +57,12 @@ local metadata_schema = {
     properties = {
         log_format = {
             type = "object"
-        }
+        },
+        max_pending_entries = {
+            type = "integer",
+            description = "maximum number of pending entries in the batch processor",
+            minimum = 1,
+        },
     },
 }
 
@@ -117,9 +123,12 @@ end
 
 
 function _M.log(conf, ctx)
+    local metadata = plugin.plugin_metadata(plugin_name)
+    local max_pending_entries = metadata and metadata.value and
+                                metadata.value.max_pending_entries or nil
     local entry = log_util.get_log_entry(plugin_name, conf, ctx)
 
-    if batch_processor_manager:add_entry(conf, entry) then
+    if batch_processor_manager:add_entry(conf, entry, max_pending_entries) then
         return
     end
 
@@ -139,7 +148,7 @@ function _M.log(conf, ctx)
         return send_udp_data(conf, data)
     end
 
-    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func)
+    batch_processor_manager:add_entry_to_new_processor(conf, entry, ctx, func, max_pending_entries)
 end
 
 return _M
diff --git a/docs/en/latest/plugins/clickhouse-logger.md b/docs/en/latest/plugins/clickhouse-logger.md
index 023f9e920bcd..0d96f6c05f70 100644
--- a/docs/en/latest/plugins/clickhouse-logger.md
+++ b/docs/en/latest/plugins/clickhouse-logger.md
@@ -104,6 +104,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/elasticsearch-logger.md b/docs/en/latest/plugins/elasticsearch-logger.md
index 80ea0760163e..8c6ecbfb9d84 100644
--- a/docs/en/latest/plugins/elasticsearch-logger.md
+++ b/docs/en/latest/plugins/elasticsearch-logger.md
@@ -62,6 +62,7 @@ This Plugin supports using batch processors to aggregate and process entries (lo
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
 | log_format | object | False |  | Custom log format in key-value pairs in JSON format. Support [APISIX variables](../apisix-variable.md) and [NGINX variables](http://nginx.org/en/docs/varindex.html) in values. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 ## Examples
 
diff --git a/docs/en/latest/plugins/google-cloud-logging.md b/docs/en/latest/plugins/google-cloud-logging.md
index 85b972381c5c..2fda43784fcb 100644
--- a/docs/en/latest/plugins/google-cloud-logging.md
+++ b/docs/en/latest/plugins/google-cloud-logging.md
@@ -91,6 +91,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/http-logger.md b/docs/en/latest/plugins/http-logger.md
index d07375b0bcea..df423a226d8b 100644
--- a/docs/en/latest/plugins/http-logger.md
+++ b/docs/en/latest/plugins/http-logger.md
@@ -105,6 +105,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/kafka-logger.md b/docs/en/latest/plugins/kafka-logger.md
index a1a717c5e95e..bb75e766c023 100644
--- a/docs/en/latest/plugins/kafka-logger.md
+++ b/docs/en/latest/plugins/kafka-logger.md
@@ -139,6 +139,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/loki-logger.md b/docs/en/latest/plugins/loki-logger.md
index 65a0a86d7e74..22dade6b1678 100644
--- a/docs/en/latest/plugins/loki-logger.md
+++ b/docs/en/latest/plugins/loki-logger.md
@@ -68,6 +68,7 @@ You can also configure log format on a global scale using the [Plugin Metadata](
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
 | log_format | object | False |  | Custom log format in key-value pairs in JSON format. Support [APISIX variables](../apisix-variable.md) and [NGINX variables](http://nginx.org/en/docs/varindex.html) in values. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 ## Examples
 
diff --git a/docs/en/latest/plugins/rocketmq-logger.md b/docs/en/latest/plugins/rocketmq-logger.md
index ff09c668eb1c..1f329c6d3f81 100644
--- a/docs/en/latest/plugins/rocketmq-logger.md
+++ b/docs/en/latest/plugins/rocketmq-logger.md
@@ -182,6 +182,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                    |
 |------------|--------|----------|-------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | log_format | object | False    |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/skywalking-logger.md b/docs/en/latest/plugins/skywalking-logger.md
index bb0532a2a72a..4541ca5fc2e6 100644
--- a/docs/en/latest/plugins/skywalking-logger.md
+++ b/docs/en/latest/plugins/skywalking-logger.md
@@ -62,6 +62,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |  | Custom log format in key-value pairs in JSON format. Support [APISIX](../apisix-variable.md) or [NGINX variables](http://nginx.org/en/docs/varindex.html) in values. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 ## Examples
 
diff --git a/docs/en/latest/plugins/splunk-hec-logging.md b/docs/en/latest/plugins/splunk-hec-logging.md
index 565f84e4bc67..a6bb44dbbc2d 100644
--- a/docs/en/latest/plugins/splunk-hec-logging.md
+++ b/docs/en/latest/plugins/splunk-hec-logging.md
@@ -87,6 +87,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/tcp-logger.md b/docs/en/latest/plugins/tcp-logger.md
index 53fd43e6609f..514538452365 100644
--- a/docs/en/latest/plugins/tcp-logger.md
+++ b/docs/en/latest/plugins/tcp-logger.md
@@ -100,6 +100,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/tencent-cloud-cls.md b/docs/en/latest/plugins/tencent-cloud-cls.md
index f1ee4c43d7d9..c02ab10f1341 100644
--- a/docs/en/latest/plugins/tencent-cloud-cls.md
+++ b/docs/en/latest/plugins/tencent-cloud-cls.md
@@ -99,6 +99,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |   | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/en/latest/plugins/udp-logger.md b/docs/en/latest/plugins/udp-logger.md
index 503566a07fd5..8001974c26af 100644
--- a/docs/en/latest/plugins/udp-logger.md
+++ b/docs/en/latest/plugins/udp-logger.md
@@ -98,6 +98,7 @@ You can also set the format of the logs by configuring the Plugin metadata. The
 | Name       | Type   | Required | Default                                                                       | Description                                                                                                                                                                                                                                             |
 | ---------- | ------ | -------- | ----------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | log_format | object | False    |  | Log format declared as key value pairs in JSON format. Values only support strings. [APISIX](../apisix-variable.md) or [Nginx](http://nginx.org/en/docs/varindex.html) variables can be used by prefixing the string with `$`. |
+| max_pending_entries | integer | False | | Maximum number of pending entries that can be buffered in batch processor before it starts dropping them. |
 
 :::info IMPORTANT
 
diff --git a/docs/zh/latest/plugins/clickhouse-logger.md b/docs/zh/latest/plugins/clickhouse-logger.md
index f482f449d362..05cf471b33c4 100644
--- a/docs/zh/latest/plugins/clickhouse-logger.md
+++ b/docs/zh/latest/plugins/clickhouse-logger.md
@@ -104,6 +104,7 @@ description: 本文介绍了 API 网关 Apache APISIX 如何使用 clickhouse-lo
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否   |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头，则表明是要获取 [APISIX](../apisix-variable.md) 或 [NGINX](http://nginx.org/en/docs/varindex.html) 变量。该配置全局生效。如果你指定了 `log_format`，该配置就会对所有绑定 `clickhouse-logger` 的路由或服务生效。|
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::note
 
diff --git a/docs/zh/latest/plugins/elasticsearch-logger.md b/docs/zh/latest/plugins/elasticsearch-logger.md
index 362b579593b0..af473dff60e6 100644
--- a/docs/zh/latest/plugins/elasticsearch-logger.md
+++ b/docs/zh/latest/plugins/elasticsearch-logger.md
@@ -63,6 +63,7 @@ description: elasticsearch-logger Plugin 将请求和响应日志批量推送到
 | Name | Type | Required | Default | Description |
 |------|------|----------|---------|-------------|
 | log_format | object | 否 |  |自定义日志格式为 JSON 格式的键值对。值中支持 [APISIX 变量](../apisix-variable.md) 和 [NGINX 变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 ## 示例
 
diff --git a/docs/zh/latest/plugins/google-cloud-logging.md b/docs/zh/latest/plugins/google-cloud-logging.md
index d485bee31009..58c05dc1321a 100644
--- a/docs/zh/latest/plugins/google-cloud-logging.md
+++ b/docs/zh/latest/plugins/google-cloud-logging.md
@@ -89,6 +89,7 @@ description: API 网关 Apache APISIX 的 google-cloud-logging 插件可用于
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否    |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头。则表明获取 [APISIX 变量](../apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::info 注意
 
diff --git a/docs/zh/latest/plugins/http-logger.md b/docs/zh/latest/plugins/http-logger.md
index 1903ddd8b134..5bde3434c032 100644
--- a/docs/zh/latest/plugins/http-logger.md
+++ b/docs/zh/latest/plugins/http-logger.md
@@ -99,6 +99,7 @@ description: 本文介绍了 API 网关 Apache APISIX 的 http-logger 插件。
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否    |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头。则表明获取 [APISIX 变量](../../../en/latest/apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::info 注意
 
diff --git a/docs/zh/latest/plugins/kafka-logger.md b/docs/zh/latest/plugins/kafka-logger.md
index e708a21b87e5..886140e11b82 100644
--- a/docs/zh/latest/plugins/kafka-logger.md
+++ b/docs/zh/latest/plugins/kafka-logger.md
@@ -135,6 +135,7 @@ description: API 网关 Apache APISIX 的 kafka-logger 插件用于将日志作
 | 名称             | 类型    | 必选项 | 默认值        |  描述                                             |
 | ---------------- | ------- | ------ | ------------- |------------------------------------------------ |
 | log_format       | object  | 否   |   | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头，则表明是要获取 [APISIX 变量](../../../en/latest/apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::note 注意
 
diff --git a/docs/zh/latest/plugins/loki-logger.md b/docs/zh/latest/plugins/loki-logger.md
index bc32dcad1179..3294c8969953 100644
--- a/docs/zh/latest/plugins/loki-logger.md
+++ b/docs/zh/latest/plugins/loki-logger.md
@@ -68,6 +68,7 @@ description: loki-logger 插件通过 Loki HTTP API /loki/api/v1/push 将请求
 | 名称 | 类型 | 必选项 | 默认值 | 描述 |
 |------|------|----------|--|-------------|
 | log_format | object | 否 |  | 日志格式以 JSON 格式声明为键值对。值只支持字符串类型。可以通过在字符串前面加上 `$` 来使用 [APISIX 变量](../apisix-variable.md) 和 [NGINX 变量](http://nginx.org/en/docs/varindex.html) 。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 ## 示例
 
diff --git a/docs/zh/latest/plugins/rocketmq-logger.md b/docs/zh/latest/plugins/rocketmq-logger.md
index 21d8e42845e5..dae7854f4645 100644
--- a/docs/zh/latest/plugins/rocketmq-logger.md
+++ b/docs/zh/latest/plugins/rocketmq-logger.md
@@ -123,6 +123,7 @@ description: API 网关 Apache APISIX 的 rocketmq-logger 插件用于将日志
 | 名称         | 类型     | 必选项 | 默认值                                                                           | 描述                                                                                                                                                               |
 |------------|--------|-----|-------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | log_format | object | 否   |  | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头，则表明是要获取 [APISIX 变量](../../../en/latest/apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::note 注意
 
diff --git a/docs/zh/latest/plugins/skywalking-logger.md b/docs/zh/latest/plugins/skywalking-logger.md
index 79eab655ed6a..e1231da839ea 100644
--- a/docs/zh/latest/plugins/skywalking-logger.md
+++ b/docs/zh/latest/plugins/skywalking-logger.md
@@ -61,6 +61,7 @@ description: skywalking-logger 将请求和响应日志作为 JSON 对象批量
 | 名称                    | 类型    | 必选项 | 默认值                | 有效值        | 描述                                                               |
 | ---------------------- | ------- | ------ | -------------------- | ------------- | ---------------------------------------------------------------- |
 | log_format | object | 否    |  | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头，则表明是要获取 [APISIX 变量](../apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 ## 示例
 
diff --git a/docs/zh/latest/plugins/splunk-hec-logging.md b/docs/zh/latest/plugins/splunk-hec-logging.md
index ef22594096e1..7b7983b02d10 100644
--- a/docs/zh/latest/plugins/splunk-hec-logging.md
+++ b/docs/zh/latest/plugins/splunk-hec-logging.md
@@ -84,6 +84,7 @@ description: API 网关 Apache APISIX 的 splunk-hec-logging 插件可用于将
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否    |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头。则表明获取 [APISIX 变量](../apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::info 注意
 
diff --git a/docs/zh/latest/plugins/tcp-logger.md b/docs/zh/latest/plugins/tcp-logger.md
index 158bb7cba293..b787c16bc5ef 100644
--- a/docs/zh/latest/plugins/tcp-logger.md
+++ b/docs/zh/latest/plugins/tcp-logger.md
@@ -95,6 +95,7 @@ description: 本文介绍了 API 网关 Apache APISIX 如何使用 tcp-logger 
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否    |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头。则表明获取 [APISIX 变量](../apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::info 注意
 
diff --git a/docs/zh/latest/plugins/tencent-cloud-cls.md b/docs/zh/latest/plugins/tencent-cloud-cls.md
index a5c41bd72d62..418d371bd491 100644
--- a/docs/zh/latest/plugins/tencent-cloud-cls.md
+++ b/docs/zh/latest/plugins/tencent-cloud-cls.md
@@ -97,6 +97,7 @@ description: API 网关 Apache APISIX tencent-cloud-cls 插件可用于将日志
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否    |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头。则表明获取 [APISIX 变量](../../../en/latest/apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::info 重要
 
diff --git a/docs/zh/latest/plugins/udp-logger.md b/docs/zh/latest/plugins/udp-logger.md
index 45d0983ffbde..d0efa2a08487 100644
--- a/docs/zh/latest/plugins/udp-logger.md
+++ b/docs/zh/latest/plugins/udp-logger.md
@@ -94,6 +94,7 @@ description: 本文介绍了 API 网关 Apache APISIX 如何使用 udp-logger 
 | 名称             | 类型    | 必选项 | 默认值        | 有效值  | 描述                                             |
 | ---------------- | ------- | ------ | ------------- | ------- | ------------------------------------------------ |
 | log_format       | object  | 否    |  |         | 以 JSON 格式的键值对来声明日志格式。对于值部分，仅支持字符串。如果是以 `$` 开头。则表明获取 [APISIX 变量](../apisix-variable.md) 或 [NGINX 内置变量](http://nginx.org/en/docs/varindex.html)。 |
+| max_pending_entries | integer | 否 | | | 在批处理器中开始删除待处理条目之前可以购买的最大待处理条目数。|
 
 :::info 注意
 
diff --git a/t/plugin/clickhouse-logger.t b/t/plugin/clickhouse-logger.t
index 4efcf11e3d70..e3f90c339b05 100644
--- a/t/plugin/clickhouse-logger.t
+++ b/t/plugin/clickhouse-logger.t
@@ -313,3 +313,90 @@ GET /opentracing
 echo "select * from default.test" | curl 'http://localhost:8123/' --data-binary @-
 --- response_body_like
 .*127.0.0.1.*1.*
+
+
+
+=== TEST 12: should drop entries when max_pending_entries is exceeded
+--- extra_yaml_config
+plugins:
+  - clickhouse-logger
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["clickhouse-logger"] = {
+                            user = "default",
+                            password = "a",
+                            database = "default",
+                            logtable = "t",
+                            endpoint_addr = "http://127.0.0.1:1234/clickhouse-logger/test1",
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/clickhouse-logger', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
diff --git a/t/plugin/elasticsearch-logger2.t b/t/plugin/elasticsearch-logger2.t
new file mode 100644
index 000000000000..18610a3e6f8f
--- /dev/null
+++ b/t/plugin/elasticsearch-logger2.t
@@ -0,0 +1,121 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 t::APISIX 'no_plan';
+
+log_level('debug');
+repeat_each(1);
+no_long_string();
+no_root_location();
+no_shuffle();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!defined $block->request) {
+        $block->set_value("request", "GET /t");
+    }
+
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: should drop entries when max_pending_entries is exceededA
+--- extra_yaml_config
+plugins:
+  - elasticsearch-logger
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["elasticsearch-logger"] = {
+                            endpoint_addr = "http://127.0.0.1:1234",
+                            field = {
+                                index = "services"
+                            },
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/elasticsearch-logger', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
diff --git a/t/plugin/google-cloud-logging3.t b/t/plugin/google-cloud-logging3.t
new file mode 100644
index 000000000000..df70f5de5a72
--- /dev/null
+++ b/t/plugin/google-cloud-logging3.t
@@ -0,0 +1,153 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!defined $block->request) {
+        $block->set_value("request", "GET /t");
+    }
+
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: should drop entries when max_pending_entries is exceeded
+--- extra_yaml_config
+plugins:
+  - google-cloud-logging
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["google-cloud-logging"] = {
+                            auth_config = {
+                                client_email = "email@apisix.iam.gserviceaccount.com",
+                                private_key = [[
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDDzrFwnA3EvYyR
+aeMgaLD3hBjvxKrz10uox1X8q7YYhf2ViRtLRUMa2bEMYksE5hbhwpNf6mKAnLOC
+UuAT6cPPdUl/agKpJXviBPIR2LuzD17WsLJHp1HxUDssSkgfCaGcOGGNfLUhhIpF
+2JUctLmxiZoAZySlSjcwupSuDJ0aPm0XO8r9H8Qu5kF2Vkz5e5bFivLTmvzrQTe4
+v5V1UI6hThElCSeUmdNF3uG3wopxlvq4zXgLTnuLbrNf/Gc4mlpV+UDgTISj32Ep
+AB2vxKEbvQw4ti8YJnGXWjxLerhfrszFw+V8lpeduiDYA44ZFoVqvzxeIsVZNtcw
+Iu7PvEPNAgMBAAECggEAVpyN9m7A1F631/aLheFpLgMbeKt4puV7zQtnaJ2XrZ9P
+PR7pmNDpTu4uF3k/D8qrIm+L+uhVa+hkquf3wDct6w1JVnfQ93riImbnoKdK13ic
+DcEZCwLjByfjFMNCxZ/gAZca55fbExlqhFy6EHmMjhB8s2LsXcTHRuGxNI/Vyi49
+sxECibe0U53aqdJbVWrphIS67cpwl4TUkN6mrHsNuDYNJ9dgkpapoqp4FTFQsBqC
+afOK5qgJ68dWZ47FBUng+AZjdCncqAIuJxxItGVQP6YPsFs+OXcivIVHJr363TpC
+l85FfdvqWV5OGBbwSKhNwiTNUVvfSQVmtURGWG/HbQKBgQD4gZ1z9+Lx19kT9WTz
+lw93lxso++uhAPDTKviyWSRoEe5aN3LCd4My+/Aj+sk4ON/s2BV3ska5Im93j+vC
+rCv3uPn1n2jUhWuJ3bDqipeTW4n/CQA2m/8vd26TMk22yOkkqw2MIA8sjJ//SD7g
+tdG7up6DgGMP4hgbO89uGU7DAwKBgQDJtkKd0grh3u52Foeh9YaiAgYRwc65IE16
+UyD1OJxIuX/dYQDLlo5KyyngFa1ZhWIs7qC7r3xXH+10kfJY+Q+5YMjmZjlL8SR1
+Ujqd02R9F2//6OeswyReachJZbZdtiEw3lPa4jVFYfhSe0M2ZPxMwvoXb25eyCNI
+1lYjSKq87wKBgHnLTNghjeDp4UKe6rNYPgRm0rDrhziJtX5JeUov1mALKb6dnmkh
+GfRK9g8sQqKDfXwfC6Z2gaMK9YaryujGaWYoCpoPXtmJ6oLPXH4XHuLh4mhUiP46
+xn8FEfSimuQS4/FMxH8A128GHQSI7AhGFFzlwfrBWcvXC+mNDsTvMmLxAoGARc+4
+upppfccETQZ7JsitMgD1TMwA2f2eEwoWTAitvlXFNT9PYSbYVHaAJbga6PLLCbYF
+FzAjHpxEOKYSdEyu7n/ayDL0/Z2V+qzc8KarDsg/0RgwppBbU/nUgeKb/U79qcYo
+y4ai3UKNCS70Ei1dTMvmdpnwXwlxfNIBufB6dy0CgYBMYq9Lc31GkC6PcGEEbx6W
+vjImOadWZbuOVnvEQjb5XCdcOsWsMcg96PtoeuyyHmhnEF1GsMzcIdQv/PHrvYpK
+Yp8D0aqsLEgwGrJQER26FPpKmyIwvcL+nm6q5W31PnU9AOC/WEkB6Zs58hsMzD2S
+kEJQcmfVew5mFXyxuEn3zA==
+-----END PRIVATE KEY-----]],
+                                project_id = "apisix",
+                                token_uri = "http://127.0.0.1:1234/google/logging/token",
+                                scope = {
+                                    "https://apisix.apache.org/logs:admin"
+                                },
+                                entries_uri = "http://127.0.0.1:1234/google/logging/entries",
+                            },
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/google-cloud-logging', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
diff --git a/t/plugin/http-logger3.t b/t/plugin/http-logger3.t
new file mode 100644
index 000000000000..70557e40e277
--- /dev/null
+++ b/t/plugin/http-logger3.t
@@ -0,0 +1,121 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 t::APISIX 'no_plan';
+
+log_level('debug');
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!$block->request) {
+        $block->set_value("request", "GET /t");
+    }
+    my $extra_init_by_lua = <<_EOC_;
+    local bpm = require("apisix.utils.batch-processor-manager")
+    bpm.set_check_stale_interval(1)
+_EOC_
+
+    $block->set_value("extra_init_by_lua", $extra_init_by_lua);
+});
+
+run_tests;
+
+__DATA__
+
+=== TEST 1: should drop entries when max_pending_entries is exceeded
+--- extra_yaml_config
+plugins:
+  - http-logger
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["http-logger"] = {
+                            uri = "http://127.0.0.1:1234/http-logger/test",
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/http-logger', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
diff --git a/t/plugin/loki-logger2.t b/t/plugin/loki-logger2.t
new file mode 100644
index 000000000000..a0f0f0ab9e12
--- /dev/null
+++ b/t/plugin/loki-logger2.t
@@ -0,0 +1,114 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!defined $block->request) {
+        $block->set_value("request", "GET /t");
+    }
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: should drop entries when max_pending_entries is exceeded
+--- extra_yaml_config
+plugins:
+  - loki-logger
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["loki-logger"] = {
+                            endpoint_addrs = {"http://127.0.0.1:1234"},
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/loki-logger', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
diff --git a/t/plugin/skywalking-logger2.t b/t/plugin/skywalking-logger2.t
new file mode 100644
index 000000000000..0148ec4fcbbb
--- /dev/null
+++ b/t/plugin/skywalking-logger2.t
@@ -0,0 +1,114 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 t::APISIX 'no_plan';
+
+log_level('debug');
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+    if (!$block->request) {
+        $block->set_value("request", "GET /t");
+    }
+});
+
+run_tests;
+
+__DATA__
+
+=== TEST 1: should drop entries when max_pending_entries is exceededA
+--- extra_yaml_config
+plugins:
+  - skywalking-logger
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["skywalking-logger"] = {
+                            endpoint_addr = "http://127.0.0.1:1234/v3/logs",
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/skywalking-logger', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5
diff --git a/t/plugin/splunk-hec-logging2.t b/t/plugin/splunk-hec-logging2.t
new file mode 100644
index 000000000000..d9d4a40184c2
--- /dev/null
+++ b/t/plugin/splunk-hec-logging2.t
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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 t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!defined $block->request) {
+        $block->set_value("request", "GET /t");
+    }
+
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: should drop entries when max_pending_entries is exceeded
+--- extra_yaml_config
+plugins:
+  - splunk-hec-logging
+--- config
+location /t {
+    content_by_lua_block {
+        local http = require "resty.http"
+        local httpc = http.new()
+        local data = {
+            {
+                input = {
+                    plugins = {
+                        ["splunk-hec-logging"] = {
+                            endpoint = {
+                                uri = "http://127.0.0.1:1234/services/collector",
+                                token = "BD274822-96AA-4DA6-90EC-18940FB2414C"
+                            },
+                            batch_max_size = 1,
+                            timeout = 1,
+                            max_retry_count = 10
+                        }
+                    },
+                    upstream = {
+                        nodes = {
+                            ["127.0.0.1:1980"] = 1
+                        },
+                        type = "roundrobin"
+                    },
+                    uri = "/hello",
+                },
+            },
+        }
+
+        local t = require("lib.test_admin").test
+
+        -- Set plugin metadata
+        local metadata = {
+            log_format = {
+                host = "$host",
+                ["@timestamp"] = "$time_iso8601",
+                client_ip = "$remote_addr"
+            },
+            max_pending_entries = 1
+        }
+
+        local code, body = t('/apisix/admin/plugin_metadata/splunk-hec-logging', ngx.HTTP_PUT, metadata)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        -- Create route
+        local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, data[1].input)
+        if code >= 300 then
+            ngx.status = code
+            ngx.say(body)
+            return
+        end
+
+        local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/hello"
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        httpc:request_uri(uri, {
+            method = "GET",
+            keepalive_timeout = 1,
+            keepalive_pool = 1,
+        })
+        ngx.sleep(2)
+    }
+}
+--- error_log
+max pending entries limit exceeded. discarding entry
+--- timeout: 5