'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):

    if ((self.config['api_version'] >= (0, 8, 1)) and (self.config['group_id'] is not None)):
        self._coordinator.refresh_committed_offsets_if_needed()
    self._fetcher.update_fetch_positions(partitions)
