rdkafka2, rdkafka: A producer created during shutdown is never closed, and the worker waits for message.timeout.ms on exit #625

Describe the bug

rdkafka2 and rdkafka close their producers in shutdown, but Fluentd stops the flush threads later, in after_shutdown. A flush thread can still call write while shutdown_producers is running. If it does, get_producer creates a new producer after shutdown_producers has cleared @producers. The plugin never closes this producer.

With rdkafka 0.30.0 or later, at_exit { Rdkafka::Clients.close_all } closes it on exit with close and no timeout. With the patch in lib/fluent/plugin/rdkafka_patch/0_16_0.rb, close without a timeout waits for the polling thread with no time limit. If the brokers can't be reached, the polling thread ends only when the queued messages expire. So the worker waits for message.timeout.ms (300 seconds by default) before it exits.

This happens with the default settings when the brokers can't be reached. The first write waits 30 seconds for the delivery report (rdkafka_delivery_handle_poll_timeout) and fails. Its retry comes while shutdown_producers waits up to 10 seconds in producer.close(10), and the retry creates the new producer.

On master, the worker crashes with SIGSEGV or hangs before this point (#611). With the fix for #611 (#624), it stops, but only after about 5 minutes.

To Reproduce

1. Use fluent-plugin-kafka with the fix for rdkafka2: Worker crashes with SIGSEGV or hangs on shutdown when messages are still queued #611 (out_rdkafka, out_rdkafka2: Fix crash or hang on shutdown with queued messages #624), Fluentd 1.20.1, and rdkafka 0.31.0.
2. Run Fluentd with the configuration below. Nothing listens on 127.0.0.1:1.
3. After 5 seconds, send SIGTERM to the supervisor.

Expected behavior

The worker stops soon after the "Queue is forcefully closed after 10 seconds wait" warning.

Your Environment

- Fluentd version: 1.20.1
- TD Agent version: N/A
- fluent-plugin-kafka version: master (5e8272a) with the fix for #611 (#624)
- rdkafka version: 0.31.0 (librdkafka 2.15.1)
- Ruby version: 4.0.9
- Operating system: CachyOS
- Kernel version: 7.3.3-1-cachyos

Your Configuration

<source>
  @type sample
  sample {"hello":"world"}
  tag test
  rate 1
</source>

<match test>
  @type rdkafka2
  brokers 127.0.0.1:1
  default_topic test
  <format>
    @type json
  </format>
  <buffer>
    @type memory
    flush_interval 1s
  </buffer>
</match>

Your Error Log

I logged when producers were created and closed. The worker stopped 338 seconds after SIGTERM.

17:40:12.011 producer created 1496 in flush_thread_0
[info]: #0 fluentd worker is now stopping worker=0
[warn]: #0 failed to flush the buffer. retry_times=0 next_retry_time=2026-11-12 17:40:43 +0900 ...
[info]: #0 shutting down output plugin type=:rdkafka2 plugin_id="object:570"
17:40:42.423 producer close 1496 args=[10]
17:40:52.424 producer close done 1496 => false
[warn]: #0 Queue is forcefully closed after 10 seconds wait
17:40:52.425 producer created 1520 in flush_thread_0
[warn]: #0 failed to flush the buffer. retry_times=1 next_retry_time=2026-11-12 17:41:24 +0900 ...
17:41:24.376 producer close 1520 args=[]
17:45:52.612 producer close done 1520 => true
[info]: Worker 0 finished with status 0

The second producer was created right after the first one was closed. It was closed only on exit, and the close finished 300 seconds after the producer was created.

Additional context

rdkafka has the same problem. I checked it with rdkafka_delivery_handle_poll_timeout 2. After the first producer was closed, the flush thread created a second one, and the worker didn't stop for 100 seconds.

share_producer true of rdkafka2 doesn't have this problem, because get_producer doesn't create a new shared producer.

One way to fix this is to close the producers after the flush threads have stopped, for example in close.

This was found while reviewing the fix for #611.