out_forward: fix error of ack handling conflict on stopping with require_ack_response enabled

[daipom]

Which issue(s) this PR fixes:
Fixes #3962

What this PR does / why we need it:
out_forward has a thread always checking the ack and closing the socket.

• fluentd/lib/fluent/plugin/out_forward.rb

  Line 512 in e89092c

  ack_check(select_interval)

When stopping Fluentd, some other threads try to do it at the same time.

• fluentd/lib/fluent/plugin/out_forward.rb

  Line 347 in e89092c

  last_ack if @require_ack_response

• fluentd/lib/fluent/plugin/out_forward.rb

  Line 376 in e89092c

  last_ack if @require_ack_response && @suspend_flush

This somtimes causes errors.
Although these errors are harmless (at least as far as I can confirm), we should fix this.

This fixes 2 points.

1. Suppress the error log.
   • When another thread closes the socket, there is no need to output an error log.
2. Fix an unhandled error of ForwardOutput::ach_check().
   • Consideration of the possibility that args of the block function of AckHandler::collect_response() can be nil was lacking from before.

The logs will change as follows.

• before:

[info]: #0 fluent/log.rb:330:info: shutting down output plugin type=:forward plugin_id="object:ba4"
[info]: #0 fluent/log.rb:330:info: delayed_commit_timeout is overwritten by ack_response_timeout
[error]: #0 fluent/log.rb:372:error: unexpected error while receiving ack message error_class=IOError error="stream closed in another thread"
[error]: #0 fluent/log.rb:372:error: unexpected error while receiving ack error_class=NoMethodError error="undefined method `disable!' for nil:NilClass"
[info]: fluent/log.rb:330:info: Worker 0 finished with status 0

• after

[info]: #0 fluent/log.rb:330:info: shutting down output plugin type=:forward plugin_id="object:ba4"
[info]: #0 fluent/log.rb:330:info: delayed_commit_timeout is overwritten by ack_response_timeout
[info]: #0 fluent/log.rb:330:info: socket closed while receiving ack response
[info]: fluent/log.rb:330:info: Worker 0 finished with status 0

Docs Changes:
Not needed.

Release Note:
Same with the title.

[daipom]

The test added in this PR seems to be unstable. I will check it.

[daipom]

https://github.com/fluent/fluentd/actions/runs/4023398771/jobs/6914222464#step:6:45

I checked this failing test and I found another conflict that we should take care of.
The conflict scenario is as follows.

• Multiple threads succeed in sock.recv() to the same socket at the same time.

  • fluentd/lib/fluent/plugin/out_forward/ack_handler.rb

    Line 115 in c26fc27

    raw_data = sock.instance_of?(Fluent::PluginHelper::Socket::WrappedSocket::TLS) ? sock.readpartial(@read_length) : sock.recv(@read_length)

• Then, those threads try to find the socket info by find().

  • fluentd/lib/fluent/plugin/out_forward/ack_handler.rb

    Line 123 in c26fc27

    info = find(sock)

• However, the socket info can be deleted in delete().

  • fluentd/lib/fluent/plugin/out_forward/ack_handler.rb

    Line 150 in c26fc27

    delete(info)

• So, if the socket info is deleted by another thread during sock.recv() and find(), the result of find() can be nil.
• This causes the unexpected error.

[daipom]

I have made additional fixes.
The current failing tests seem to have nothing to do with this fix.

[ashie]

LGTM
I commented a nitpick.

[ashie]

Thanks!

[daipom]

Thanks for your review!