Need authentication when sending tcp heartbeat and keepalive in out_forward

lib/fluent/plugin/out_forward.rb

@@ -578,12 +578,7 @@ def standby?
 
       def verify_connection
         connect do |sock, ri|
-          if ri.state != :established
-            establish_connection(sock, ri)
-            if ri.state != :established
-              raise "Failed to establish connection to #{@host}:#{@port}"
-            end
-          end
+          ensure_established_connection(sock, ri)
         end
       end
 
@@ -652,14 +647,7 @@ def send_data_actual(sock, tag, chunk)
       def send_data(tag, chunk)
         ack = @ack_handler && @ack_handler.create_ack(chunk.unique_id, self)
         connect(nil, ack: ack) do |sock, ri|
-          if ri.state != :established
-            establish_connection(sock, ri)
-
-            if ri.state != :established
-              raise ConnectionClosedError, "failed to establish connection with node #{@name}"
-            end
-          end
-
+          ensure_established_connection(sock, ri)
           send_data_actual(sock, tag, chunk)
         end
 
@@ -684,7 +672,9 @@ def send_heartbeat
 
         case @sender.heartbeat_type
         when :transport
-          connect(dest_addr) do |_ri, _sock|
+          connect(dest_addr) do |sock, ri|
+            ensure_established_connection(sock, ri)
+
             ## don't send any data to not cause a compatibility problem
             # sock.write FORWARD_TCP_HEARTBEAT_DATA
 
@@ -776,6 +766,16 @@ def heartbeat(detect=true)
 
       private
 
+      def ensure_established_connection(sock, request_info)
+        if request_info.state != :established
+          establish_connection(sock, request_info)
+
+          if request_info.state != :established
+            raise ConnectionClosedError, "failed to establish connection with node #{@name}"
+          end
+        end
+      end
+
       def connect(host = nil, ack: false, &block)
         @connection_manager.connect(host: host || resolved_host, port: port, hostname: @hostname, ack: ack, &block)
       end

test/plugin/test_out_forward.rb

@@ -1056,7 +1056,7 @@ def plugin_id_for_test?
         e = assert_raise Fluent::UnrecoverableError do
           d.instance_start
         end
-        assert_match(/Failed to establish connection/, e.message)
+        assert_match(/failed to establish connection/, e.message)
       end
     end
 
@@ -1092,7 +1092,7 @@ def plugin_id_for_test?
           d.instance_start
         end
 
-        assert_match(/Failed to establish connection/, e.message)
+        assert_match(/failed to establish connection/, e.message)
       end
     end