out_forward: handle closed ACK sockets gracefully

In the threaded ack receiving case, we must be sure to check
IOError when doing socket operation, in order not to crash on
socket closed by another thread.

Signed-off-by: Fujimoto Seiji <[email protected]>

lib/fluent/plugin/out_forward/ack_handler.rb

@@ -59,7 +59,13 @@ def collect_response(select_interval)
             @ack_waitings = new_list
           end
 
-          readable_sockets, _, _ = IO.select(sockets, nil, nil, select_interval)
+          begin
+            readable_sockets, _, _ = IO.select(sockets, nil, nil, select_interval)
+          rescue IOError
+            @log.info "connection closed while waiting for readable sockets"
+            readable_sockets = nil
+          end
+
           if readable_sockets
             readable_sockets.each do |sock|
               results << read_ack_from_sock(sock)
@@ -109,6 +115,9 @@ def read_ack_from_sock(sock)
           raw_data = sock.instance_of?(Fluent::PluginHelper::Socket::WrappedSocket::TLS) ? sock.readpartial(@read_length) : sock.recv(@read_length)
         rescue Errno::ECONNRESET, EOFError # ECONNRESET for #recv, #EOFError for #readpartial
           raw_data = ''
+        rescue IOError
+          @log.info "socket closed while receiving ack response"
+          return nil, Result::FAILED
         end
 
         info = find(sock)