test: in_forward: Add skipping invalid condition to assert encapsulated timestamps

Signed-off-by: Hiroshi Hatake <hiroshi@chronosphere.io>

Author: cosmo0920

test/plugin/test_in_forward.rb

@@ -277,7 +277,7 @@ def create_driver(conf=base_config)
     end
 
     test 'metadata attached logs type in entries array' do
-      @d = d = create_driver
+      @d = d = create_driver(base_config + "skip_invalid_event false")
 
       time = event_time("2011-01-02 13:14:15 UTC")
       metadata = {}
@@ -292,6 +292,7 @@ def create_driver(conf=base_config)
         send_data packer.write(["tag1", entries, {"fluent_signal" => 0}]).to_s
       end
       assert_equal(["tag1", "tag1"], d.events.map { |tag, _time, _record| tag })
+      assert_equal([time, time], d.events.map { |_tag, time, _record| time })
       assert_equal(records, d.events.map { |_tag, _time, record| record })
     end
 
@@ -427,7 +428,7 @@ def create_driver(conf=base_config)
     end
 
     test 'metadata attached logs type in packed entries' do
-      @d = d = create_driver
+      @d = d = create_driver(base_config + "skip_invalid_event false")
 
       time = event_time("2011-01-02 13:14:15 UTC")
       metadata = {}
@@ -445,6 +446,7 @@ def create_driver(conf=base_config)
         send_data packer.write(["tag1", entries, {"fluent_signal" => 0}]).to_s
       end
       assert_equal(["tag1", "tag1"], d.events.map { |tag, _time, _record| tag })
+      assert_equal([time, time], d.events.map { |_tag, time, _record| time })
       assert_equal(records, d.events.map { |_tag, _time, record| record })
     end
 
@@ -550,6 +552,33 @@ def create_driver(conf=base_config)
   end
 
   sub_test_case 'compressed packed forward' do
+    test 'metadata attached logs type without skip_invalid_event' do
+      @d = d = create_driver(base_config + "skip_invalid_event false")
+
+      time = event_time("2011-01-02 13:14:15 UTC")
+      metadata = {}
+      records = [
+        {"a"=>1},
+        {"a"=>2},
+      ]
+
+      entries = ''
+      records.each do |record|
+        entries << compress([[time, metadata], record].to_msgpack)
+      end
+      chunk = ["tag1", entries, { 'compressed' => 'gzip', 'fluent_signal' => 0 }].to_msgpack
+
+      d.run do
+        Fluent::MessagePackFactory.msgpack_unpacker.feed_each(chunk) do |obj|
+          d.instance.send(:on_message, obj, chunk.size, DUMMY_SOCK)
+        end
+      end
+
+      assert_equal(["tag1", "tag1"], d.events.map { |tag, _time, _record| tag })
+      assert_equal([time, time], d.events.map { |_tag, _time, _record| _time })
+      assert_equal(records, d.events.map { |_tag, _time, record| record })
+    end
+
     test 'set_compress_to_option_gzip' do
       @d = d = create_driver