Critical bug fix: Extra sanity checking when marking offsets as processed

lib/kafka/offset_manager.rb

@@ -50,9 +50,20 @@ def set_default_offset(topic, default_offset)
     # @param offset [Integer] the offset of the message that should be marked as processed.
     # @return [nil]
     def mark_as_processed(topic, partition, offset)
-      @uncommitted_offsets += 1
+      unless @group.assigned_to?(topic, partition)
+        @logger.debug "Not marking #{topic}/#{partition}:#{offset} as processed for partition not assigned to this consumer."
+        return
+      end
       @processed_offsets[topic] ||= {}
 
+      last_processed_offset = @processed_offsets[topic][partition] || -1
+      if last_processed_offset > offset + 1
+        @logger.debug "Not overwriting newer offset #{topic}/#{partition}:#{last_processed_offset - 1} with older #{offset}"
+        return
+      end
+
+      @uncommitted_offsets += 1
+
       # The committed offset should always be the offset of the next message that the
       # application will read, thus adding one to the last message processed.
       @processed_offsets[topic][partition] = offset + 1

spec/offset_manager_spec.rb

@@ -21,9 +21,13 @@
   }
   let(:offset_retention_time) { nil }
   let(:commit_interval) { 0 }
+  let(:partition_assignments) { { 'greetings' => [0, 1, 2] } }
 
   before do
     allow(group).to receive(:commit_offsets)
+    allow(group).to receive(:assigned_to?) do |topic, partition|
+      (partition_assignments[topic] || []).include?(partition)
+    end
     allow(fetcher).to receive(:seek)
   end
 
@@ -43,6 +47,46 @@
 
       expect(group).to have_received(:commit_offsets).with(expected_offsets)
     end
+
+    context "after calling #mark_as_processed with offsets from non-assigned partitions" do
+      it "only commits offsets from assigned partitions" do
+        offset_manager.mark_as_processed("greetings", 0, 42)
+        offset_manager.mark_as_processed("greetings", 1, 13)
+        offset_manager.mark_as_processed("greetings", 5, 75)
+        offset_manager.mark_as_processed("seasons-greetings", 3, 15)
+
+        offset_manager.commit_offsets
+
+        expected_offsets = {
+          "greetings" => {
+            0 => 43,
+            1 => 14,
+          }
+        }
+
+        expect(group).to have_received(:commit_offsets).with(expected_offsets)
+      end
+    end
+
+    context "after marking offsets as processed for the same partition but out of order" do
+      it "committs the newest offset" do
+        offset_manager.mark_as_processed("greetings", 0, 42)
+        offset_manager.mark_as_processed("greetings", 1, 579)
+        offset_manager.mark_as_processed("greetings", 0, 5)
+        offset_manager.mark_as_processed("greetings", 1, 95)
+
+        offset_manager.commit_offsets
+
+        expected_offsets = {
+          "greetings" => {
+            0 => 43,
+            1 => 580
+          }
+        }
+
+        expect(group).to have_received(:commit_offsets).with(expected_offsets)
+      end
+    end
   end
 
   describe "#commit_offsets_if_necessary" do
@@ -192,6 +236,7 @@ def partition_offset_info(offset)
   end
 
   describe "#clear_offsets_excluding" do
+    let(:partition_assignments) { { 'x' => [0, 1] } }
     it "clears offsets except for the partitions in the exclusion list" do
       offset_manager.mark_as_processed("x", 0, 42)
       offset_manager.mark_as_processed("x", 1, 13)