def commit_offsets(recommit = false)
      offsets = offsets_to_commit(recommit)
      unless offsets.empty?
        @logger.debug "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
def commit_offsets(recommit = false)
      offsets = offsets_to_commit(recommit)
      unless offsets.empty?
        @logger.debug "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
Commit offsets of messages that have been marked as processed.

 If `recommit` is set to true, we will also commit the existing positions
 even if no messages have been processed on a partition. This is done
 in order to avoid the offset information expiring in cases where messages
 are very rare -- it's essentially a keep-alive.

 @param recommit [Boolean] whether to recommit offsets that have already been
   committed.
 @return [nil]