Fix the cancellation flow to avoid multiple invocations of basic.cancel

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 7.3.3
+ - Fixed the cancellation flow to avoid multiple invocations of basic.cancel [#55](https://github.com/logstash-plugins/logstash-integration-rabbitmq/pull/55)
+
 ## 7.3.2
  - Change `tls_certificate_password` type to `password` to protect from leaks in the logs [#54](https://github.com/logstash-plugins/logstash-integration-rabbitmq/pull/54)
 

lib/logstash/inputs/rabbitmq.rb

@@ -308,7 +308,14 @@ def stop
  end
 
  def shutdown_consumer
- return unless @consumer
+ # There are two possible flows to shutdown consumers. When the plugin is the one shutting down, it should send a channel
+ # cancellation message by invoking channel.basic_cancel(consumer_tag) and waiting for the consumer to terminate
+ # (broker replies with an basic.cancel-ok). This back and forth is handled by MarchHare. On the other hand, when the broker
+ # requests the client to shutdown (eg. due to queue deletion). It sends to the client a basic.cancel message, which is handled
+ # internally by the client, unregistering the consumer and then invoking the :on_cancellation callback. In that case, the plugin
+ # should not do anything as the consumer is already cancelled/unregistered.
+ return if !@consumer || @consumer.cancelled? || @consumer.terminated?
+
  @hare_info.channel.basic_cancel(@consumer.consumer_tag)
  connection = @hare_info.connection
  until @consumer.terminated?

logstash-integration-rabbitmq.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
  s.name = 'logstash-integration-rabbitmq'
- s.version = '7.3.2'
+ s.version = '7.3.3'
  s.licenses = ['Apache License (2.0)']
  s.summary = "Integration with RabbitMQ - input and output plugins"
  s.description = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline "+

spec/inputs/rabbitmq_spec.rb

@@ -63,6 +63,62 @@
  expect(instance.codec.config_name).to eq "json"
  end
 
+ describe "and #stop is called" do
+ let(:consumer_tag) { "foo-bar-queue" }
+ let(:consumer) { double("consumer") }
+
+ before do
+ instance.register
+ instance.setup!
+ instance.instance_variable_set(:@consumer, consumer)
+
+ allow(instance).to receive(:close_connection)
+ allow(consumer).to receive(:consumer_tag).and_return(consumer_tag)
+ end
+
+ context "with a cancelled consumer" do
+ before do
+ allow(consumer).to receive(:cancelled?).and_return(true)
+ allow(consumer).to receive(:terminated?).and_return(false)
+ end
+
+ it "should not call basic_cancel" do
+ expect(channel).to_not receive(:basic_cancel)
+ instance.stop
+ end
+ end
+
+ context "with a terminated consumer" do
+ before do
+ allow(consumer).to receive(:cancelled?).and_return(false)
+ allow(consumer).to receive(:terminated?).and_return(true)
+ end
+
+ it "should not call basic_cancel" do
+ expect(channel).to_not receive(:basic_cancel)
+ instance.stop
+ end
+ end
+
+ context "with a running consumer" do
+ before do
+ allow(consumer).to receive(:cancelled?).and_return(false)
+ allow(consumer).to receive(:terminated?).and_return(false, false, true)
+ end
+
+ it "should call basic_cancel" do
+ expect(channel).to receive(:basic_cancel).with(consumer_tag)
+ instance.stop
+ end
+
+ it "should log terminating info" do
+ allow(channel).to receive(:basic_cancel).with(consumer_tag)
+ expect(instance.logger).to receive(:info).with(/Waiting for RabbitMQ consumer to terminate before stopping/, anything)
+ instance.stop
+ end
+ end
+ end
+
  describe "#connect!" do
  subject { hare_info }