fix(batch-processor): we didn't free stale object actually (apache#5700)

apisix/plugins/http-logger.lua

@@ -25,6 +25,7 @@ local plugin          = require("apisix.plugin")
 local ngx      = ngx
 local tostring = tostring
 local ipairs   = ipairs
+local pairs    = pairs
 local timer_at = ngx.timer.at
 
 local plugin_name = "http-logger"
@@ -150,7 +151,7 @@ local function remove_stale_objects(premature)
         return
     end
 
-    for key, batch in ipairs(buffers) do
+    for key, batch in pairs(buffers) do
         if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
             core.log.warn("removing batch processor stale object, conf: ",
                           core.json.delay_encode(key))

apisix/plugins/kafka-logger.lua

@@ -23,7 +23,6 @@ local plugin = require("apisix.plugin")
 local math     = math
 local pairs    = pairs
 local type     = type
-local ipairs   = ipairs
 local plugin_name = "kafka-logger"
 local stale_timer_running = false
 local timer_at = ngx.timer.at
@@ -138,7 +137,7 @@ local function remove_stale_objects(premature)
         return
     end
 
-    for key, batch in ipairs(buffers) do
+    for key, batch in pairs(buffers) do
         if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
             core.log.warn("removing batch processor stale object, conf: ",
                           core.json.delay_encode(key))

apisix/plugins/sls-logger.lua

@@ -26,6 +26,7 @@ local tcp = ngx.socket.tcp
 local buffers = {}
 local tostring = tostring
 local ipairs = ipairs
+local pairs = pairs
 local table = table
 local schema = {
     type = "object",
@@ -116,7 +117,7 @@ local function remove_stale_objects(premature)
         return
     end
 
-    for key, batch in ipairs(buffers) do
+    for key, batch in pairs(buffers) do
         if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
             core.log.warn("removing batch processor stale object, route id:", tostring(key))
             buffers[key] = nil

apisix/plugins/syslog.lua

@@ -22,7 +22,7 @@ local logger_socket = require("resty.logger.socket")
 local plugin_name = "syslog"
 local ngx = ngx
 local buffers = {}
-local ipairs   = ipairs
+local pairs = pairs
 local stale_timer_running = false;
 local timer_at = ngx.timer.at
 
@@ -121,7 +121,7 @@ local function remove_stale_objects(premature)
         return
     end
 
-    for key, batch in ipairs(buffers) do
+    for key, batch in pairs(buffers) do
         if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
             core.log.warn("removing batch processor stale object, conf: ",
                           core.json.delay_encode(key))

apisix/plugins/tcp-logger.lua

@@ -22,7 +22,7 @@ local tostring = tostring
 local buffers = {}
 local ngx = ngx
 local tcp = ngx.socket.tcp
-local ipairs   = ipairs
+local pairs = pairs
 local stale_timer_running = false
 local timer_at = ngx.timer.at
 
@@ -106,7 +106,7 @@ local function remove_stale_objects(premature)
         return
     end
 
-    for key, batch in ipairs(buffers) do
+    for key, batch in pairs(buffers) do
         if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
             core.log.warn("removing batch processor stale object, conf: ",
                           core.json.delay_encode(key))

apisix/plugins/udp-logger.lua

@@ -22,7 +22,7 @@ local tostring = tostring
 local buffers = {}
 local ngx = ngx
 local udp = ngx.socket.udp
-local ipairs   = ipairs
+local pairs = pairs
 local stale_timer_running = false;
 local timer_at = ngx.timer.at
 
@@ -90,7 +90,7 @@ local function remove_stale_objects(premature)
         return
     end
 
-    for key, batch in ipairs(buffers) do
+    for key, batch in pairs(buffers) do
         if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
             core.log.warn("removing batch processor stale object, conf: ",
                           core.json.delay_encode(key))