'Update the fetch positions for the provided partitions.
Arguments:
partitions (list of TopicPartitions): partitions to update
Raises:
NoOffsetForPartitionError: if no offset is stored for a given
partition and no reset policy is available'

def update_fetch_positions(self, partitions):

    for tp in partitions:
        if (not self._subscriptions.is_assigned(tp)):
            log.warning('partition    %s    is    not    assigned    -    skipping    offset    update', tp)
            continue
        elif self._subscriptions.is_fetchable(tp):
            log.warning('partition    %s    is    still    fetchable    --    skipping    offset    update', tp)
            continue
        if self._subscriptions.is_offset_reset_needed(tp):
            self._reset_offset(tp)
        elif (self._subscriptions.assignment[tp].committed is None):
            self._subscriptions.need_offset_reset(tp)
            self._reset_offset(tp)
        else:
            committed = self._subscriptions.assignment[tp].committed
            log.debug('Resetting    offset    for    partition    %s    to    the    committed    offset    %s', tp, committed)
            self._subscriptions.seek(tp, committed)
