Backport(v1.16): server plugin helper: ensure to close all connections at shutdown (#5026) (#5088)

**Which issue(s) this PR fixes**:
* Backport #5026

**What this PR does / why we need it**:
TCP server with `server` helper does not close all connections at
shutdown process.
When receiving data from multiple clients, the server receive the data
continuously because the connection is not closed.

The server will shut down properly by this PR

### Reproduce
1. Launch Fluentd with following config file.
2. Send syslog data from two or more clients using following client
script
3. Terminate Fluentd
4. Relaunch Fluentd, then it shows the `2025-07-16 14:12:26 +0900
[warn]: #0 restoring buffer file: path = xxxxxxxxx` in logs.

* config

```
<source>
  @type syslog
  tag system

  <transport tcp>
  </transport>

  bind 0.0.0.0
  port 5140
</source>

<match **>
  @type file
  path "#{File.expand_path('~/tmp/fluentd/maillog')}"
  <buffer>
    @type file
    path "#{File.expand_path('~/tmp/fluentd/buffer/buffer_syslog_maillog')}"
    flush_at_shutdown true
  </buffer>
</match>
```

* client script

```ruby
require 'bundler/inline'
gemfile do
  source 'https://rubygems.org'
  gem 'remote_syslog_sender'
end

def create_client
  Thread.new do
    sender = RemoteSyslogSender.new('127.0.0.1', 5140, protocol: :tcp)
    loop do
      sender.transmit("message body")
      sleep 0.5
    end
  end
end

clients = []
3.times do
  clients << create_client
end

clients.each(&:join)
```

Example code.
```
@ary = [1,2,3]

# It would like 1, 2, and 3 to be processed.
# However, following code handles 1 and 3.
@ary.each do |i|
  puts i
  @ary.delete(i)
end
```

**Docs Changes**:
Not needed.

**Release Note**:
Same as the title.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Signed-off-by: Daijiro Fukuda <fukuda@clear-code.com>
Co-authored-by: Shizuo Fujita <fujita@clear-code.com>

Author: daipom

lib/fluent/plugin_helper/server.rb

@@ -347,7 +347,10 @@ def stop
       end
 
       def shutdown
-        @_server_connections.each do |conn|
+        # When it invokes conn.cose, it reduces elements in @_server_connections by close_callback,
+        # and it reduces the number of loops. This prevents the connection closing.
+        # So, it requires invoking #dup to avoid the problem.
+        @_server_connections.dup.each do |conn|
           conn.close rescue nil
         end
 

test/plugin_helper/test_server.rb

@@ -364,6 +364,25 @@ class Dummy < Fluent::Plugin::TestBase
         d2.stop; d2.before_shutdown; d2.shutdown; d2.after_shutdown; d2.close; d2.terminate
       end
     end
+
+    test 'close all connections by shutdown' do
+      @d.server_create_tcp(:s, @port) do |data, conn|
+      end
+
+      client_sockets = []
+      5.times do
+        client_sockets << TCPSocket.open("127.0.0.1", @port)
+      end
+      waiting(4){ sleep 0.1 until @d.instance_variable_get(:@_server_connections).size == 5 }
+
+      @d.stop
+      @d.before_shutdown
+      @d.shutdown
+
+      assert_true @d.instance_variable_get(:@_server_connections).empty?
+    ensure
+      client_sockets.each(&:close)
+    end
   end
 
   sub_test_case '#server_create' do