[Chore] Try to debug on CircleCI

Author: lmduc

lib/kafka/consumer.rb

@@ -469,6 +469,7 @@ def make_final_offsets_commit!(attempts = 3)
     end
 
     def join_group
+      puts "Join group"
       old_generation_id = @group.generation_id
 
       @group.join
@@ -490,6 +491,7 @@ def join_group
 
       @fetcher.reset
 
+      puts "assigned_partitions: #{@group.assigned_partitions}"
       @group.assigned_partitions.each do |topic, partitions|
         partitions.each do |partition|
           if paused?(topic, partition)
@@ -499,9 +501,11 @@ def join_group
           end
         end
       end
+      puts "finished join group"
     end
 
     def seek_to_next(topic, partition)
+      puts "Seek to #{topic} - #{partition}"
       # When automatic marking is off, the first poll needs to be based on the last committed
       # offset from Kafka, that's why we fallback in case of nil (it may not be 0)
       if @current_offsets[topic].key?(partition)
@@ -534,6 +538,7 @@ def fetch_batches
       # Return early if the consumer has been stopped.
       return [] if shutting_down?
 
+      puts "@group.member #{@group.member?}"
       join_group unless @group.member?
 
       trigger_heartbeat

lib/kafka/consumer_group.rb

@@ -41,19 +41,23 @@ def member?
     end
 
     def join
+      puts "Consumer group joining"
       if @topics.empty?
         raise Kafka::Error, "Cannot join group without at least one topic subscription"
       end
 
       join_group
       synchronize
+      puts "Consumer group joined"
     rescue NotCoordinatorForGroup
       @logger.error "Failed to find coordinator for group `#{@group_id}`; retrying..."
+      puts "Failed to find coordinator for group `#{@group_id}`; retrying..."
       sleep 1
       @coordinator = nil
       retry
     rescue ConnectionError
       @logger.error "Connection error while trying to join group `#{@group_id}`; retrying..."
+      puts "Connection error while trying to join group `#{@group_id}`; retrying..."
       sleep 1
       @cluster.mark_as_stale!
       @coordinator = nil
@@ -174,6 +178,7 @@ def synchronize
       if group_leader?
         @logger.info "Chosen as leader of group `#{@group_id}`"
 
+        puts "assignment strategy #{@members.keys} - #{@topics}"
         group_assignment = @assignment_strategy.assign(
           members: @members.keys,
           topics: @topics,
@@ -192,6 +197,7 @@ def synchronize
 
         response.member_assignment.topics.each do |topic, assigned_partitions|
           @logger.info "Partitions assigned for `#{topic}`: #{assigned_partitions.join(', ')}"
+          puts "Partitions assigned for `#{topic}`: #{assigned_partitions.join(', ')}"
         end
 
         @assigned_partitions.replace(response.member_assignment.topics)

spec/functional/consumer_group_spec.rb

@@ -123,6 +123,7 @@
 
     producer = Kafka.new(kafka_brokers, client_id: "test").producer
 
+    puts "Starting pushing a"
     messages_a.each { |i| producer.produce(i.to_s, topic: topic_a) }
     producer.deliver_messages
 
@@ -145,6 +146,7 @@
     end
     thread.abort_on_exception = true
 
+    puts "Starting pushing b"
     messages_b.each { |i| producer.produce(i.to_s, topic: topic_b) }
     producer.deliver_messages