out_forward errors on shutdown with require_ack_response enabled

[fujimotos]

Describe the bug

When we stop td-agent service, out_forward sometimes throws
an unhandled exception.

The exact exception varies, but it always occures in ack_handler.rb.
Here is a few examples:

# case 1
unexpected error while receiving ack error_class=IOError
error="closed stream"

# case 2
unexpected error while receiving ack message error_class=IOError
error="stream closed in another thread"

# case 3
unexpected error while receiving ack error_class=NoMethodError
error="undefined method `disable!' for nil:NilClass"

It suggests that out_forward's shutdown sequence is racy i.e.
multiple threads can process the same ack message concurrently.

To Reproduce

• Run out_foward with require_ack_response
• Stop Fluentd during actively transmitting records.

Expected behavior

No error.

Your Environment

- Fluentd version: v1.15.2
- TD Agent version: -
- Operating system: -
- Kernel version: -

Your Configuration

<match **>
  @type forward
  require_ack_response true
  ...
</match>

Your Error Log

See above.

Additional context

No response

[daipom]

I note how to reproduce this.

• Launch 2 Fleuntd (Aggregator and Forwarder)

Config of Aggregator

<source>
  @type forward
</source>

<filter test.**>
  @type record_transformer
  enable_ruby
  <record>
    hoge ${ sleep(1); "hoge" } # Insert `sleep` to stack the ack check
  </record>
</filter>

<match test.**>
  @type stdout
</match>

Config of Forwarder

<source>
  @type sample
  tag test
</source>

<match test.**>
  @type forward
  require_ack_response true
  <buffer tag,time,file>
    @type file
    path /test/fluentd/forwarder/buffer
    timekey 24h
    flush_mode immediate
    flush_at_shutdown true
  </buffer>
  <server>
    host localhost
    port 24224
  </server>
</match>

• Wait a while and we can see some chunk files staying in the Forwarder.
• Then stop the Forwarder and check the log of Fluentd.

This is a little difficult.
It doesn't seem that storing many chunks makes it easier to reproduce.
Increasing the sleep time seems to make it harder.
It may depend on the machine specs, but if I stop Fluentd when 2 chunk files accumulate, it is easy to reproduce.