'Update offsets using auto_offset_reset policy (smallest|largest)
Arguments:
partition (int): the partition for which offsets should be updated
Returns: Updated offset on success, None on failure'

def reset_partition_offset(self, partition):

    LATEST = (-1)
    EARLIEST = (-2)
    if (self.auto_offset_reset == 'largest'):
        reqs = [OffsetRequestPayload(self.topic, partition, LATEST, 1)]
    elif (self.auto_offset_reset == 'smallest'):
        reqs = [OffsetRequestPayload(self.topic, partition, EARLIEST, 1)]
    else:
        if (sys.exc_info() == (None, None, None)):
            raise OffsetOutOfRangeError('Cannot    reset    partition    offsets    without    a    valid    auto_offset_reset    setting    (largest|smallest)')
        raise
    log.info('Resetting    topic-partition    offset    to    %s    for    %s:%d', self.auto_offset_reset, self.topic, partition)
    try:
        (resp,) = self.client.send_offset_request(reqs)
    except KafkaError as e:
        log.error('%s    sending    offset    request    for    %s:%d', e.__class__.__name__, self.topic, partition)
    else:
        self.offsets[partition] = resp.offsets[0]
        self.fetch_offsets[partition] = resp.offsets[0]
        return resp.offsets[0]
