'Seek to the oldest available offset for partitions.
Arguments:
*partitions: Optionally provide specific TopicPartitions, otherwise
default to all assigned partitions.
Raises:
AssertionError: If any partition is not currently assigned, or if
no partitions are assigned.'

def seek_to_beginning(self, *partitions):

    if (not all([isinstance(p, TopicPartition) for p in partitions])):
        raise TypeError('partitions    must    be    TopicPartition    namedtuples')
    if (not partitions):
        partitions = self._subscription.assigned_partitions()
        assert partitions, 'No    partitions    are    currently    assigned'
    else:
        for p in partitions:
            assert (p in self._subscription.assigned_partitions()), 'Unassigned    partition'
    for tp in partitions:
        log.debug('Seeking    to    beginning    of    partition    %s', tp)
        self._subscription.need_offset_reset(tp, OffsetResetStrategy.EARLIEST)
