Merge pull request #1 from catawiki/sav_include_clean_offsets

include clean offsets on stable 0.6

Author: sarcilav

lib/kafka/consumer.rb

@@ -439,13 +439,15 @@ def join_group
       if old_generation_id && @group.generation_id != old_generation_id + 1
         # We've been out of the group for at least an entire generation, no
         # sense in trying to hold on to offset data
+        clear_current_offsets
         @offset_manager.clear_offsets
       else
         # After rejoining the group we may have been assigned a new set of
         # partitions. Keeping the old offset commits around forever would risk
         # having the consumer go back and reprocess messages if it's assigned
         # a partition it used to be assigned to way back. For that reason, we
         # only keep commits for the partitions that we're still assigned.
+        clear_current_offsets(excluding: @group.assigned_partitions)
         @offset_manager.clear_offsets_excluding(@group.assigned_partitions)
       end
 
@@ -532,5 +534,13 @@ def fetch_batches
     def pause_for(topic, partition)
       @pauses[topic][partition]
     end
+
+    def clear_current_offsets(excluding: {})
+      @current_offsets.each do |topic, partitions|
+        partitions.keep_if do |partition, _|
+          excluding.fetch(topic, []).include?(partition)
+        end
+      end
+    end
   end
 end

lib/kafka/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module Kafka
-  VERSION = "0.6.8"
+  VERSION = "0.6.10-beta1"
 end