[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

lib/kafka/fetcher.rb

@@ -109,6 +109,7 @@ def loop
 
         @logger.debug "Handling fetcher command: #{cmd}"
 
+        puts "Handling fetcher command: #{cmd} - #{args}"
         send("handle_#{cmd}", *args)
       elsif @queue.size < @max_queue_size
         step
@@ -158,6 +159,7 @@ def handle_seek(topic, partition, offset)
 
     def step
       batches = fetch_batches
+      puts "Fetched #{batches.size} messages"
 
       batches.each do |batch|
         unless batch.empty?
@@ -197,6 +199,7 @@ def fetch_batches
         max_bytes = @max_bytes_per_partition[topic]
 
         partitions.each do |partition, offset|
+          puts "Fetching from #{topic} - partition #{partition} - offset #{offset}"
           operation.fetch_from_partition(topic, partition, offset: offset, max_bytes: max_bytes)
         end
       end

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