ISSUE-525/764: Local @topics variable does not contain all the subscription information

The leader consumer thread needs to have knowledge about
all siblings consumer subscriptions in order to be perform
the correct partition assignments.
For consumers with different subscription this is not possible
through the local `@topics`. A way to solve this is by obtaining
this info from the cluster itself through the `@members` var.

Author: Eduardo Poleo

lib/kafka/consumer_group.rb

@@ -190,9 +190,14 @@ def synchronize
       if group_leader?
         @logger.info "Chosen as leader of group `#{@group_id}`"
 
+        topics = Set.new
+        @members.each do |_member, metadata|
+          metadata.topics.each { |t| topics.add(t) }
+        end
+
         group_assignment = @assignor.assign(
           members: @members,
-          topics: @topics,
+          topics: topics,
         )
       end
 

spec/functional/consumer_group_spec.rb

@@ -518,6 +518,68 @@ def call(cluster:, members:, partitions:)
     expect(received_messages.values.map(&:count)).to match_array [messages.count / 3, messages.count / 3 * 2]
   end
 
+  example "subscribing to different topics while in the same consumer group" do
+    topic1 = create_random_topic(num_partitions: 1)
+    topic2 = create_random_topic(num_partitions: 1)
+    messages = (1..500).to_a
+
+    begin
+      kafka = Kafka.new(kafka_brokers, client_id: "test")
+      producer = kafka.producer
+
+      messages[0..249].each do |i|
+        producer.produce(i.to_s, topic: topic1, partition: 0)
+      end
+
+      messages[250..500].each do |i|
+        producer.produce(i.to_s, topic: topic2, partition: 0)
+      end
+
+      producer.deliver_messages
+    end
+
+    group_id = "test#{rand(1000)}"
+
+    mutex = Mutex.new
+    received_messages = []
+
+    assignment_strategy_class = Kafka::MultiSubscriptionRoundRobinAssignmentStrategy
+
+    consumers = [topic1, topic2].map do |topic|
+      assignment_strategy = assignment_strategy_class.new
+      kafka = Kafka.new(kafka_brokers, client_id: "test", logger: logger)
+      consumer = kafka.consumer(
+        group_id: group_id,
+        offset_retention_time: offset_retention_time,
+        assignment_strategy: assignment_strategy
+      )
+      consumer.subscribe(topic)
+      consumer
+    end
+
+    threads = consumers.map do |consumer|
+      t = Thread.new do
+        consumer.each_message do |message|
+          mutex.synchronize do
+            received_messages << message
+
+            if received_messages.count == messages.count
+              consumers.each(&:stop)
+            end
+          end
+        end
+      end
+
+      t.abort_on_exception = true
+
+      t
+    end
+
+    threads.each(&:join)
+
+    expect(received_messages.map(&:value).map(&:to_i)).to match_array messages
+  end
+
   def wait_until(timeout:)
     Timeout.timeout(timeout) do
       sleep 0.5 until yield