stop/shutdown_consumer can throw on exit, leading to upstream crashes #45
Description
When exiting the rabbitmq plugin, if the connection was already closed for some reason, the shutdown_consumer method called by stop can throw an exception, leading to a fatal crash of the Logstash process.
Logstash information:
Observed on logstash 7.10.1, plugin version 7.1.1
Steps to reproduce:
This behavior was observed, while Logstash was being shut down -
[2021-02-02T11:41:21,337][ERROR][logstash.agent ] Failed to execute action {:action=>LogStash::PipelineAction::Stop/pipeline_id:mgsf5, :exception=>"Java::ComRabbitmqClient::AlreadyClosedException", :message=>"connection is already closed due to connection error; cause: com.rabbitmq.client.MissedHeartbeatException: Heartbeat missing with heartbeat = 60 seconds", :backtrace=>["com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(com/rabbitmq/client/impl/AMQChannel.java:258)", "com.rabbitmq.client.impl.AMQChannel.rpc(com/rabbitmq/client/impl/AMQChannel.java:341)", "com.rabbitmq.client.impl.ChannelN.basicCancel(com/rabbitmq/client/impl/ChannelN.java:1491)", "jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)", "jdk.internal.reflect.NativeMethodAccessorImpl.invoke(jdk/internal/reflect/NativeMethodAccessorImpl.java:62)", "jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(jdk/internal/reflect/DelegatingMethodAccessorImpl.java:43)", "java.lang.reflect.Method.invoke(java/lang/reflect/Method.java:566)", "org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(org/jruby/javasupport/JavaMethod.java:426)", "org.jruby.javasupport.JavaMethod.invokeDirect(org/jruby/javasupport/JavaMethod.java:293)", "org.jruby.RubyClass.finvokeWithRefinements(org/jruby/RubyClass.java:514)", "org.jruby.RubyBasicObject.send(org/jruby/RubyBasicObject.java:1755)", "org.jruby.RubyBasicObject$INVOKER$i$send.call(org/jruby/RubyBasicObject$INVOKER$i$send.gen)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.march_hare_minus_4_dot_3_dot_0_minus_java.lib.march_hare.channel.method_missing(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/march_hare-4.3.0-java/lib/march_hare/channel.rb:888)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_rabbitmq_minus_7_dot_1_dot_1_minus_java.lib.logstash.inputs.rabbitmq.shutdown_consumer(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-rabbitmq-7.1.1-java/lib/logstash/inputs/rabbitmq.rb:318)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_rabbitmq_minus_7_dot_1_dot_1_minus_java.lib.logstash.inputs.rabbitmq.RUBY$method$shutdown_consumer$0$__VARARGS__(usr/share/logstash/vendor/bundle/jruby/$2_dot_5_dot_0/gems/logstash_minus_integration_minus_rabbitmq_minus_7_dot_1_dot_1_minus_java/lib/logstash/inputs//usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-rabbitmq-7.1.1-java/lib/logstash/inputs/rabbitmq.rb)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_rabbitmq_minus_7_dot_1_dot_1_minus_java.lib.logstash.inputs.rabbitmq.stop(/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-rabbitmq-7.1.1-java/lib/logstash/inputs/rabbitmq.rb:312)", "usr.share.logstash.vendor.bundle.jruby.$2_dot_5_dot_0.gems.logstash_minus_integration_minus_rabbitmq_minus_7_dot_1_dot_1_minus_java.lib.logstash.inputs.rabbitmq.RUBY$method$stop$0$__VARARGS__(usr/share/logstash/vendor/bundle/jruby/$2_dot_5_dot_0/gems/logstash_minus_integration_minus_rabbitmq_minus_7_dot_1_dot_1_minus_java/lib/logstash/inputs//usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-rabbitmq-7.1.1-java/lib/logstash/inputs/rabbitmq.rb)", "usr.share.logstash.logstash_minus_core.lib.logstash.inputs.base.do_stop(/usr/share/logstash/logstash-core/lib/logstash/inputs/base.rb:103)", "usr.share.logstash.logstash_minus_core.lib.logstash.inputs.base.RUBY$method$do_stop$0$__VARARGS__(usr/share/logstash/logstash_minus_core/lib/logstash/inputs//usr/share/logstash/logstash-core/lib/logstash/inputs/base.rb)", "org.jruby.RubySymbol$SymbolProcBody.yieldInner(org/jruby/RubySymbol.java:1440)", "org.jruby.RubySymbol$SymbolProcBody.doYield(org/jruby/RubySymbol.java:1456)", "org.jruby.RubyArray.each(org/jruby/RubyArray.java:1809)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.stop_inputs(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:458)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$stop_inputs$0$__VARARGS__(usr/share/logstash/logstash_minus_core/lib/logstash//usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.shutdown(/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:447)", "usr.share.logstash.logstash_minus_core.lib.logstash.java_pipeline.RUBY$method$shutdown$0$__VARARGS__(usr/share/logstash/logstash_minus_core/lib/logstash//usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipeline_action.stop.execute(/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/stop.rb:30)", "RUBY.terminate_pipeline(/usr/share/logstash/logstash-core/lib/logstash/pipelines_registry.rb:173)", "usr.share.logstash.logstash_minus_core.lib.logstash.pipeline_action.stop.execute(/usr/share/logstash/logstash-core/lib/logstash/pipeline_action/stop.rb:29)", "usr.share.logstash.logstash_minus_core.lib.logstash.agent.converge_state(/usr/share/logstash/logstash-core/lib/logstash/agent.rb:365)", "org.jruby.RubyProc.call(org/jruby/RubyProc.java:318)", "java.lang.Thread.run(java/lang/Thread.java:834)"]}
[2021-02-02T11:41:21,432][INFO ][logstash.pipelinesregistry] Removed pipeline from registry successfully {:pipeline_id=>:"placement-betting"}
[2021-02-02T11:41:21,442][FATAL][logstash.runner ] An unexpected error occurred! {:error=>#<LogStash::Error: Don't know how to handle `Java::ComRabbitmqClient::AlreadyClosedException` for `PipelineAction::Stop<mgsf5>`>, :backtrace=>["org/logstash/execution/ConvergeResultExt.java:129:in `create'", "org/logstash/execution/ConvergeResultExt.java:57:in `add'", "/usr/share/logstash/logstash-core/lib/logstash/agent.rb:378:in `block in converge_state'"]}
[2021-02-02T11:41:21,452][ERROR][org.logstash.Logstash ] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit