Set the fetch position to the committed position (if there is one)
        or reset it using the offset reset policy the user has configured.

        Arguments:
            partitions (List[TopicPartition]): The partitions that need
                updating fetch positions.

        Raises:
            NoOffsetForPartitionError: If no offset is stored for a given
                partition and no offset reset policy is defined.

def _update_fetch_positions(self, partitions):
        """Set the fetch position to the committed position (if there is one)
        or reset it using the offset reset policy the user has configured.

        Arguments:
            partitions (List[TopicPartition]): The partitions that need
                updating fetch positions.

        Raises:
            NoOffsetForPartitionError: If no offset is stored for a given
                partition and no offset reset policy is defined.
        """
        # Lookup any positions for partitions which are awaiting reset (which may be the
        # case if the user called :meth:`seek_to_beginning` or :meth:`seek_to_end`. We do
        # this check first to avoid an unnecessary lookup of committed offsets (which
        # typically occurs when the user is manually assigning partitions and managing
        # their own offsets).
        self._fetcher.reset_offsets_if_needed(partitions)

        if not self._subscription.has_all_fetch_positions():
            # if we still don't have offsets for all partitions, then we should either seek
            # to the last committed position or reset using the auto reset policy
            if (self.config['api_version'] >= (0, 8, 1) and
                self.config['group_id'] is not None):
                # first refresh commits for all assigned partitions
                self._coordinator.refresh_committed_offsets_if_needed()

            # Then, do any offset lookups in case some positions are not known
            self._fetcher.update_fetch_positions(partitions)