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)