MessagePackFactory: Make sure to reset local unpacker to prevent received broken data from affecting other receiving data

lib/fluent/msgpack_factory.rb

@@ -100,7 +100,12 @@ def self.thread_local_msgpack_packer
     end
 
     def self.thread_local_msgpack_unpacker
-      Thread.current[:local_msgpack_unpacker] ||= MessagePackFactory.engine_factory.unpacker
+      unpacker = Thread.current[:local_msgpack_unpacker]
+      if unpacker.nil?
+        return Thread.current[:local_msgpack_unpacker] = MessagePackFactory.engine_factory.unpacker
+      end
+      unpacker.reset
+      unpacker
     end
   end
 end

test/test_msgpack_factory.rb

@@ -15,4 +15,36 @@ class MessagePackFactoryTest < Test::Unit::TestCase
     assert mp.msgpack_factory
     assert mp.msgpack_factory
   end
+
+  sub_test_case 'thread_local_msgpack_packer' do
+    test 'packer is cached' do
+      packer1 = Fluent::MessagePackFactory.thread_local_msgpack_packer
+      packer2 = Fluent::MessagePackFactory.thread_local_msgpack_packer
+      assert_equal packer1, packer2
+    end
+  end
+
+  sub_test_case 'thread_local_msgpack_unpacker' do
+    test 'unpacker is cached' do
+      unpacker1 = Fluent::MessagePackFactory.thread_local_msgpack_unpacker
+      unpacker2 = Fluent::MessagePackFactory.thread_local_msgpack_unpacker
+      assert_equal unpacker1, unpacker2
+    end
+
+    # We need to reset the buffer every time so that received incomplete data
+    # must not affect data from other senders.
+    test 'reset the internal buffer of unpacker every time' do
+      unpacker1 = Fluent::MessagePackFactory.thread_local_msgpack_unpacker
+      unpacker1.feed_each("\xA6foo") do |result|
+        flunk("This callback must not be called since the data is uncomplete.")
+      end
+
+      records = []
+      unpacker2 = Fluent::MessagePackFactory.thread_local_msgpack_unpacker
+      unpacker2.feed_each("\xA3foo") do |result|
+        records.append(result)
+      end
+      assert_equal ["foo"], records
+    end
+  end
 end