From 6d9eb0ec3f6ff1f929c4b179d833e8b0b6e974b4 Mon Sep 17 00:00:00 2001
From: Jonathan Gough <jonathanpgough@gmail.com>
Date: Tue, 15 Dec 2020 12:36:56 +0000
Subject: [PATCH 1/3] Decorate flushed events from multiline codecs

---
 lib/logstash/inputs/s3.rb | 34 +++++++++++++++++++---------------
 1 file changed, 19 insertions(+), 15 deletions(-)

diff --git a/lib/logstash/inputs/s3.rb b/lib/logstash/inputs/s3.rb
index b04bf74..2c03a29 100644
--- a/lib/logstash/inputs/s3.rb
+++ b/lib/logstash/inputs/s3.rb
@@ -225,30 +225,34 @@ def process_local_log(queue, filename, object)
           @logger.debug('Event is metadata, updating the current cloudfront metadata', :event => event)
           update_metadata(metadata, event)
         else
-          decorate(event)
-
-          event.set("cloudfront_version", metadata[:cloudfront_version]) unless metadata[:cloudfront_version].nil?
-          event.set("cloudfront_fields", metadata[:cloudfront_fields]) unless metadata[:cloudfront_fields].nil?
-
-          if @include_object_properties
-            event.set("[@metadata][s3]", object.data.to_h)
-          else
-            event.set("[@metadata][s3]", {})
-          end
-
-          event.set("[@metadata][s3][key]", object.key)
-
-          queue << event
+          push_decoded_event(queue, metadata, object, event)
         end
       end
     end
     # #ensure any stateful codecs (such as multi-line ) are flushed to the queue
     @codec.flush do |event|
-      queue << event
+      push_decoded_event(queue, metadata, object, event)
     end
 
     return true
   end # def process_local_log
+  
+  def push_decoded_event(queue, metadata, object, event)
+    decorate(event)
+
+    event.set("cloudfront_version", metadata[:cloudfront_version]) unless metadata[:cloudfront_version].nil?
+    event.set("cloudfront_fields", metadata[:cloudfront_fields]) unless metadata[:cloudfront_fields].nil?
+
+    if @include_object_properties
+      event.set("[@metadata][s3]", object.data.to_h)
+    else
+      event.set("[@metadata][s3]", {})
+    end
+
+    event.set("[@metadata][s3][key]", object.key)
+
+    queue << event
+  end
 
   def event_is_metadata?(event)
     return false unless event.get("message").class == String

From 446efe96dc892199cd62c5d12c88ee133a423ba6 Mon Sep 17 00:00:00 2001
From: Jonathan Gough <jonathanpgough@gmail.com>
Date: Tue, 15 Dec 2020 12:37:55 +0000
Subject: [PATCH 2/3] Test both events have metadata set correctly

---
 spec/inputs/s3_spec.rb | 1 +
 1 file changed, 1 insertion(+)

diff --git a/spec/inputs/s3_spec.rb b/spec/inputs/s3_spec.rb
index e420e0f..b4ddf1e 100644
--- a/spec/inputs/s3_spec.rb
+++ b/spec/inputs/s3_spec.rb
@@ -325,6 +325,7 @@
       events = fetch_events(config)
       expect(events.size).to eq(events_to_process)
       expect(events[0].get("[@metadata][s3][key]")).to eql log.key
+      expect(events[1].get("[@metadata][s3][key]")).to eql log.key
     end
 
     it "deletes the temporary file" do

From 3f52f4df1ad4e9c77e743394f6f64c5c33790767 Mon Sep 17 00:00:00 2001
From: Kaise Cheng <kaise.cheng@elastic.co>
Date: Thu, 20 Jan 2022 16:03:02 +0000
Subject: [PATCH 3/3] dump version

---
 CHANGELOG.md              | 3 +++
 logstash-input-s3.gemspec | 2 +-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b906dce..01e26ff 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 3.8.3
+ - Fix missing `metadata` and `type` of the last event [#223](https://github.com/logstash-plugins/logstash-input-s3/pull/223)
+
 ## 3.8.2
  - Refactor: read sincedb time once per bucket listing [#233](https://github.com/logstash-plugins/logstash-input-s3/pull/233)
 
diff --git a/logstash-input-s3.gemspec b/logstash-input-s3.gemspec
index 08f55fe..21ed2bd 100644
--- a/logstash-input-s3.gemspec
+++ b/logstash-input-s3.gemspec
@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-s3'
-  s.version         = '3.8.2'
+  s.version         = '3.8.3'
   s.licenses        = ['Apache-2.0']
   s.summary         = "Streams events from files in a S3 bucket"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"