s3 input plugin not handling shutdown correctly, leading to duplicates once started again

[PadaKwaak]

s3 input plugin not handling shutdown correctly, leading to duplicates once started again

[PadaKwaak]

In the meantime I have made a modification to the v3.4.1 on my own fork to not stop in the middle (ie. my option (b) that I suggested above), and confirmed that it is now working as expected when Logstash receives a shutdown signal while busy processing like a 6MB .gzip json encoded file:

[2021-04-16T09:50:52,717][WARN ][logstash.runner          ] SIGTERM received. Shutting down.
[2021-04-16T09:50:53,256][WARN ][org.logstash.input.DeadLetterQueueInputPlugin] closing dead letter queue input plugin
[2021-04-16T09:50:53,631][INFO ][logstash.pipeline        ] Pipeline has terminated {:pipeline_id=>"dlq-pipeline", :thread=>"#<Thread:0x3c9d0685@/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/create.rb:51 run>"}
[2021-04-16T09:50:58,965][WARN ][org.logstash.execution.ShutdownWatcherExt] {"inflight_count"=>0, "stalling_threads_info"=>{["LogStash::Filters::Grok", {"patterns_dir"=>["/opt/logstash/patterns"], "match"=>{"[@metadata][s3][key]"=>"/(?<s3_file>[^/]+)\\.gz"}, "id"=>"dd3398e8c462a0384b18a7d950b46d28cb9f7fd8956073d9e58f5f18a9ea7402"}]=>[{"thread_id"=>37, "name"=>nil, "current_call"=>"[...]/vendor/bundle/jruby/2.3.0/gems/manticore-0.6.4-java/lib/manticore/response.rb:50:in `call'"}, {"thread_id"=>38, "name"=>nil, "current_call"=>"[...]/vendor/bundle/jruby/2.3.0/gems/manticore-0.6.4-java/lib/manticore/response.rb:50:in `call'"}]}}
[2021-04-16T09:50:58,976][ERROR][org.logstash.execution.ShutdownWatcherExt] The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information.
[2021-04-16T09:51:04,004][WARN ][org.logstash.execution.ShutdownWatcherExt] {"inflight_count"=>0, "stalling_threads_info"=>{"other"=>[{"thread_id"=>39, "name"=>"[s3-pipeline]<s3", "current_call"=>"[...]/vendor/bundle/jruby/2.3.0/gems/logstash-codec-json-3.0.5/lib/logstash/codecs/json.rb:48:in `from_json'"}], ["LogStash::Filters::Grok", {"patterns_dir"=>["/opt/logstash/patterns"], "match"=>{"[@metadata][s3][key]"=>"/(?<s3_file>[^/]+)\\.gz"}, "id"=>"dd3398e8c462a0384b18a7d950b46d28cb9f7fd8956073d9e58f5f18a9ea7402"}]=>[{"thread_id"=>37, "name"=>nil, "current_call"=>"[...]/vendor/bundle/jruby/2.3.0/gems/concurrent-ruby-1.0.5-java/lib/concurrent/map.rb:93:in `[]'"}, {"thread_id"=>38, "name"=>nil, "current_call"=>"[...]/vendor/bundle/jruby/2.3.0/gems/logstash-filter-grok-4.0.4/lib/logstash/filters/grok.rb:339:in `each'"}]}}
[2021-04-16T09:51:06,799][INFO ][logstash.pipeline        ] Pipeline has terminated {:pipeline_id=>"s3-pipeline", :thread=>"#<Thread:0x748701b3 run>"}

I made the modification on top of 3.4.1, since that was the version I was using earlier: v3.4.1...PadaKwaak:v342

Once the issue with 3.6.0 has been fixed that I've raised with #225, I'll see if I can create a pull request where this above modification is enabled/disabled with a setting so that people can choose to use it or not.

[kaisecheng]

The duplication here is caused by an interruption before a file can update the checkpoint/ timestamp in sincedb. The sincedb is updated per file, so when the pipeline restarts it processes the same file again. As it does not cause a missing event, I do not classify this as a bug. A better way to handle it is enhancing sincedb to have a checkpoint per file and record the latest event id. The whole plugin is getting old and needs refactoring.

It is crucial to keep the restart process fast from the system administration point of view. I would avoid stalling the shutdown.