def position(self, partition, timeout_ms=None):
        """Get the offset of the next record that will be fetched

        Arguments:
            partition (TopicPartition): Partition to check

        Returns:
            int: Offset or None
        """
        if not isinstance(partition, TopicPartition):
            raise TypeError('partition must be a TopicPartition namedtuple')
        assert self._subscription.is_assigned(partition), 'Partition is not assigned'

        timer = Timer(timeout_ms)
        position = self._subscription.assignment[partition].position
        while position is None:
            # batch update fetch positions for any partitions without a valid position
            if self._update_fetch_positions(timeout_ms=timer.timeout_ms):
                position = self._subscription.assignment[partition].position
            elif timer.expired:
                return None
        else:
            return position.offset
Get the offset of the next record that will be fetched

        Arguments:
            partition (TopicPartition): Partition to check

        Returns:
            int: Offset or None
        
kafka/consumer/group.py