test send ack if channel is open

Author: alexcams

lib/logstash/inputs/rabbitmq.rb

@@ -59,6 +59,7 @@ module Inputs
     class RabbitMQ < LogStash::Inputs::Threadable
 
       java_import java.util.concurrent.TimeUnit
+      java_import java.util.concurrent.CountDownLatch
 
       include ::LogStash::PluginMixins::RabbitMQConnection
 
@@ -189,6 +190,7 @@ def setup!
         connect!
         declare_queue!
         bind_exchange!
+        @terminated = CountDownLatch.new(1)
         @hare_info.channel.prefetch = @prefetch_count
       rescue => e
         # when encountering an exception during shut-down,
@@ -260,15 +262,20 @@ def internal_queue_consume!
         while true
           payload = @internal_queue.poll(10, TimeUnit::MILLISECONDS)
           if !payload  # Nothing in the queue
-            if last_delivery_tag # And we have unacked stuff
+            puts "payload null, checking delivery tag"
+            if last_delivery_tag  # And we have unacked stuff
+              puts"Sending ack for last_delivery_tag #{last_delivery_tag}"
               @hare_info.channel.ack(last_delivery_tag, true) if @ack
               i=0
               last_delivery_tag = nil
             end
             next
           end
-
-          break if payload == INTERNAL_QUEUE_POISON
+          if payload == INTERNAL_QUEUE_POISON
+            puts "Received poison pill, shutting down internal queue consumer"
+            @terminated.countDown
+            break
+          end
 
           metadata, data = payload
           @codec.decode(data) do |event|
@@ -302,7 +309,12 @@ def decorate(event, metadata, data)
       end
 
       def stop
-        @internal_queue&.put(INTERNAL_QUEUE_POISON)
+        puts "Sending Poison pill"
+        @internal_queue.put(INTERNAL_QUEUE_POISON)
+        timed_out = !@terminated.await(2, TimeUnit::SECONDS)
+        if timed_out
+          @logger.warn("Timeout waiting for RabbitMQ consumer to terminate")
+        end
         shutdown_consumer
         close_connection
       end
@@ -315,7 +327,7 @@ def shutdown_consumer
         # 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?
-
+        puts "shutting down consumer"
         @hare_info.channel.basic_cancel(@consumer.consumer_tag)
         connection = @hare_info.connection
         until @consumer.terminated?

lib/logstash/plugin_mixins/rabbitmq_connection.rb

@@ -86,6 +86,7 @@ def conn_str
 
       def close_connection
         @rabbitmq_connection_stopping = true
+        puts "closing channel"
         @hare_info.channel.close if channel_open?
         @hare_info.connection.close if connection_open?
       end