[Chore] join_group need to be executed again when a new topic is added

Author: lmduc

lib/kafka/consumer.rb

@@ -77,6 +77,9 @@ def initialize(cluster:, logger:, instrumenter:, group:, fetcher:, offset_manage
       # Hash storing topics that are already being subscribed
       # When subcribing to a new topic, if it's already being subscribed before, skip it
       @subscribed_topics = Set.new
+
+      # Whether join_group must be executed again because new topics are added
+      @join_group_for_new_topics = false
     end
 
     # Subscribes the consumer to a topic.
@@ -489,6 +492,7 @@ def join_group
       end
 
       @fetcher.reset
+      @join_group_for_new_topics = false
 
       @group.assigned_partitions.each do |topic, partitions|
         partitions.each do |partition|
@@ -534,7 +538,7 @@ def fetch_batches
       # Return early if the consumer has been stopped.
       return [] if shutting_down?
 
-      join_group unless @group.member?
+      join_group if !@group.member? || @join_group_for_new_topics
 
       trigger_heartbeat
 
@@ -597,6 +601,7 @@ def subscribe_to_regex(topic_regex, default_offset, start_from_beginning, max_by
     def subscribe_to_topic(topic, default_offset, start_from_beginning, max_bytes_per_partition)
       return if @subscribed_topics.include?(topic)
       @subscribed_topics.add(topic)
+      @join_group_for_new_topics = true
 
       @group.subscribe(topic)
       @offset_manager.set_default_offset(topic, default_offset)

spec/functional/consumer_group_spec.rb

@@ -117,8 +117,8 @@
     topic_a = generate_topic_name
     topic_b = generate_topic_name
 
-    messages_a = (1..5).to_a
-    messages_b = (6..10).to_a
+    messages_a = (1..500).to_a
+    messages_b = (501..1000).to_a
     messages = messages_a + messages_b
 
     producer = Kafka.new(kafka_brokers, client_id: "test").producer