class Kafka::OffsetManager

def commit_offsets(recommit = false)
  offsets = offsets_to_commit(recommit)
  unless offsets.empty?
    @logger.info "Committing offsets#{recommit ? ' with recommit' : ''}: #{prettify_offsets(offsets)}"
    @group.commit_offsets(offsets)
    @last_commit = Time.now
    @last_recommit = Time.now if recommit
    @uncommitted_offsets = 0
    @committed_offsets = nil
  end
end