Merge pull request #3405 from ashie/fix-race-in-out-forward-handshake

out_forward: Fix a race condition on handshake
fluent · Jun 2, 2021 · 86c6a5e · 86c6a5e
2 parents 1be5323 + 9f9a67d
commit 86c6a5e
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/lib/fluent/plugin/out_forward.rb b/lib/fluent/plugin/out_forward.rb
@@ -565,8 +565,6 @@ def initialize(sender, server, failure:, connection_manager:, ack_handler:)
           username: server.username || '',
         )
 
-        @unpacker = Fluent::MessagePackFactory.msgpack_unpacker
-
         @resolved_host = nil
         @resolved_time = 0
         @resolved_once = false
@@ -613,7 +611,8 @@ def establish_connection(sock, ri)
               sleep @sender.read_interval
               next
             end
-            @unpacker.feed_each(buf) do |data|
+            unpacker = Fluent::MessagePackFactory.thread_local_msgpack_unpacker
+            unpacker.feed_each(buf) do |data|
               if @handshake.invoke(sock, ri, data) == :established
                 @log.debug "connection established", host: @host, port: @port
               end

diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
@@ -1014,7 +1014,9 @@ def try_write(chunk)
     end
 
     logs = d.logs
-    assert_false(logs.any? { |log| log.include?("invalid format for PONG message") || log.include?("shared key mismatch") }, "'#{logs.last.strip}' happens")
+    assert_false(logs.any? { |log|
+                   log.include?("invalid format for PONG message") || log.include?("shared key mismatch")
+                 }, "Actual log:\n#{logs.join}")
   end
 
   def create_target_input_driver(response_stub: nil, disconnect: false, conf: target_config)