Merge pull request #431 from zendesk/dasch/fix-offset-cache

Clear cached offset when seeking to default

Author: dasch

lib/kafka/offset_manager.rb

@@ -36,6 +36,9 @@ def mark_as_processed(topic, partition, offset)
  end
 
  def seek_to_default(topic, partition)
+ # Remove any cached offset, in case things have changed broker-side.
+ clear_resolved_offset(topic)
+
  seek_to(topic, partition, -1)
  end
 
@@ -104,6 +107,10 @@ def clear_offsets_excluding(excluded)
 
  private
 
+ def clear_resolved_offset(topic)
+ @resolved_offsets.delete(topic)
+ end
+
  def resolve_offset(topic, partition)
  @resolved_offsets[topic] ||= fetch_resolved_offsets(topic)
  @resolved_offsets[topic].fetch(partition)