feat: Raise an error when trying to use a consumer after it has been closed

Author: ukdave

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [Unreleased]
 ### Changed
+- Raise an error when trying to use a consumer after it has been closed
 - Add `Consumer#stop` method to stop an `each` loop
 - Add crystal versions 1.15.1 and 1.16.3 to test matrix
 

spec/kafka/consumer_spec.cr

@@ -25,6 +25,24 @@ describe Kafka::Consumer do
         consumer.subscribe("foo", "foo")
       end
     end
+
+    it "raises an exception when called after the consumer is closed" do
+      consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9094", "group.id" => "foo", "broker.address.family" => "v4"})
+      consumer.close
+      expect_raises(Kafka::ConsumerException, "librdkafka error - Consumer closed") do
+        consumer.subscribe("foo")
+      end
+    end
+  end
+
+  describe "#poll" do
+    it "raises an exception when called after the consumer is closed" do
+      consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9094", "group.id" => "foo", "broker.address.family" => "v4"})
+      consumer.close
+      expect_raises(Kafka::ConsumerException, "librdkafka error - Consumer closed") do
+        consumer.poll(250)
+      end
+    end
   end
 
   describe "#each" do
@@ -42,5 +60,33 @@ describe Kafka::Consumer do
         consumer.each(timeout: 10) { }
       end
     end
+
+    it "raises an exception when called after the consumer is closed" do
+      consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9094", "group.id" => "foo", "broker.address.family" => "v4"})
+      consumer.close
+      expect_raises(Kafka::ConsumerException, "librdkafka error - Consumer closed") do
+        consumer.each { }
+      end
+    end
+  end
+
+  describe "#open?" do
+    it "returns true after creation" do
+      consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9094", "group.id" => "foo", "broker.address.family" => "v4"})
+      consumer.open?.should be_true
+    end
+
+    it "returns false after closing" do
+      consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9094", "group.id" => "foo", "broker.address.family" => "v4"})
+      consumer.close
+      consumer.open?.should be_false
+    end
+  end
+
+  describe "#close" do
+    it "does not raise an exception when called multiple times" do
+      consumer = Kafka::Consumer.new({"bootstrap.servers" => "localhost:9094", "group.id" => "foo", "broker.address.family" => "v4"})
+      2.times { consumer.close }
+    end
   end
 end

src/kafka/consumer.cr

@@ -29,6 +29,7 @@ module Kafka
     #
     # Calls the `rd_kafka_subscribe` C function.
     def subscribe(*topics)
+      verify_handle_open!
       tpl = LibRdKafka.topic_partition_list_new(topics.size)
       topics.each do |topic|
         LibRdKafka.topic_partition_list_add(tpl, topic, -1)
@@ -43,6 +44,7 @@ module Kafka
     #
     # Calls the `rd_kafka_consumer_poll` C function.
     def poll(timeout_ms : Int32, raise_on_error : Bool = true) : Message?
+      verify_handle_open!
       message_ptr = LibRdKafka.consumer_poll(@handle, timeout_ms)
       return if message_ptr.null?
 
@@ -59,6 +61,7 @@ module Kafka
     #
     # At the beginning of each loop, `Fiber.yield` is called allow other Fibers to run.
     def each(timeout = 250, raise_on_error = true, &)
+      verify_handle_open!
       @running = true
       while @running
         Fiber.yield
@@ -80,13 +83,24 @@ module Kafka
       @running = false
     end
 
+    # Returns whether the consumer is open.
+    def open?
+      !@handle.null?
+    end
+
     # Close the consumer and destroy the Kafka handle.
     #
     # Calls the `rd_kafka_consumer_close` and `rd_kafka_destroy` C functions.
     def close
+      return if @handle.null?
+
       LibRdKafka.consumer_close(@handle)
       LibRdKafka.kafka_destroy(@handle)
       @handle = LibRdKafka::KafkaHandle.null
     end
+
+    private def verify_handle_open!
+      raise ConsumerException.new("Consumer closed") if @handle.null?
+    end
   end
 end